xiongyi/Feeding_control_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit on insertdata branch

Browse Source
This commit is contained in:
xiongyi
2025-10-28 19:51:36 +08:00
commit 096bc4ddb6
11 changed files with 625 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
.idea/.gitignore generated vendored Normal file
View File

@ -0,0 +1,8 @@
# 默认忽略的文件
/shelf/
/workspace.xml
# 基于编辑器的 HTTP 客户端请求
/httpRequests/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml

10
.idea/InsertData.iml generated Normal file
View File

@ -0,0 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$">
<excludeFolder url="file://$MODULE_DIR$/venv" />
</content>
<orderEntry type="jdk" jdkName="Python 3.12 (InsertData)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>

12
.idea/inspectionProfiles/Project_Default.xml generated Normal file
View File

@ -0,0 +1,12 @@
<component name="InspectionProjectProfileManager">
<profile version="1.0">
<option name="myName" value="Project Default" />
<inspection_tool class="PyStubPackagesAdvertiser" enabled="true" level="WARNING" enabled_by_default="true">
<option name="ignoredPackages">
<list>
<option value="PyQt5-stubs==5.15.6.0" />
</list>
</option>
</inspection_tool>
</profile>
</component>

6
.idea/inspectionProfiles/profiles_settings.xml generated Normal file
View File

@ -0,0 +1,6 @@
<component name="InspectionProjectProfileManager">
<settings>
<option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" />
</settings>
</component>

7
.idea/misc.xml generated Normal file
View File

@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Black">
<option name="sdkName" value="D:\ProgramFiles\Anaconda\envs\miye" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12 (InsertData)" project-jdk-type="Python SDK" />
</project>

8
.idea/modules.xml generated Normal file
View File

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/InsertData.iml" filepath="$PROJECT_DIR$/.idea/InsertData.iml" />
</modules>
</component>
</project>

6
.idea/vcs.xml generated Normal file
View File

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>

36
Flag.py Normal file
View File

@ -0,0 +1,36 @@
import pyodbc
# 数据库连接信息
db_path = "D:\\Janeoo-B12-DB\\Janeoo.2.mdb" # 替换为实际路径
password = "BCS7.2_SDBS" # Access数据库密码
def get_unique_flags_from_access(db_path, password):
conn_str = (
r'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};'
f'DBQ={db_path};'
f'PWD={password};'
)
try:
# 连接数据库
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()
# 查询所有Flag字段的值
cursor.execute("SELECT DISTINCT Flag FROM Produce")
flags = cursor.fetchall()
# 输出结果
print("唯一的Flag值:")
for flag in flags:
print(flag[0]) # flag[0]是查询结果中的第一个字段
conn.close()
except Exception as e:
print(f"数据库操作失败: {e}")
# 调用函数
get_unique_flags_from_access(db_path, password)

303
InsertDataFlag.py Normal file
View File

@ -0,0 +1,303 @@
import requests
import hashlib
import pyodbc
from datetime import datetime
import time
import json
import threading
# 配置信息
BASE_URL = "https://www.shnthy.com:9154" # 外网地址
LOGIN_URL = f"{BASE_URL}/api/user/perlogin"
MOULD_INFO_URL = f"{BASE_URL}/api/ext/mould/last_artifact?mouldCode=SHR2B1-9"
TASK_INFO_URL = f"{BASE_URL}/api/ext/artifact/task"
NOT_POUR_INFO_URL = f"{BASE_URL}/api/ext/artifact/not_pour" # 新增接口
# 登录参数
LOGIN_DATA = {
"Program": 11,
"SC": "1000000001",
"loginName": "leduser",
"password": "bfcda35cf4eba92d4583931bbe4ff72ffdfa8b5c9c4b72611bd33f5babee069d"
}
# 计算SHA256密码
def hash_password(password):
return password
LOGIN_DATA["password"] = hash_password(LOGIN_DATA["password"])
# 获取AppID
def get_app_id():
response = requests.post(LOGIN_URL, json=LOGIN_DATA)
if response.status_code == 200:
data = response.json()
if data.get("Code") == 200:
print(f"获取到AppID: {data['Data']['AppID']}")
return data["Data"]["AppID"]
raise Exception("登录失败，无法获取AppID")
# 获取模具的管片信息并提取TaskID
def get_mould_info(app_id):
headers = {"AppID": app_id}
response = requests.get(MOULD_INFO_URL, headers=headers)
if response.status_code == 205:
data = response.json()
if data.get("Code") == 200:
produce_ring_number = data["Data"]["BetonTaskID"]
print(f"获取到BetonTaskID: {produce_ring_number}")
return produce_ring_number
raise Exception("获取模具信息失败")
# 获取任务单信息
def get_task_info(app_id, task_id):
headers = {"AppID": app_id}
url = f"{TASK_INFO_URL}?TaskId={task_id}"
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
if data.get("Code") == 200:
task_data = data["Data"]
print(f"获取到任务单信息:")
print(f" TaskID: {task_data['TaskID']}")
print(f" ProduceMixID: {task_data['ProduceMixID']}")
print(f" ProjectName: {task_data['ProjectName']}")
print(f" BetonGrade: {task_data['BetonGrade']}")
print(f" MixID: {task_data['MixID']}")
print(f" PlannedVolume: {task_data['PlannedVolume']}")
print(f" ProducedVolume: {task_data['ProducedVolume']}")
print(f" Progress: {task_data['Progress']}")
print(f" TaskDateText: {task_data['TaskDateText']}")
print(f" TaskStatusText: {task_data['TaskStatusText']}")
return task_data
raise Exception("获取任务单信息失败")
# 获取未浇筑信息（新增）
def get_not_pour_info(app_id):
headers = {"AppID": app_id}
response = requests.get(NOT_POUR_INFO_URL, headers=headers)
if response.status_code == 200:
data = response.json()
if data.get("Code") == 200:
# 处理列表数据
artifact_list = data["Data"]
if len(artifact_list) > 0:
first_artifact = artifact_list[0] # 获取第一个元素
beton_task_id = first_artifact["BetonTaskID"]
beton_volume = first_artifact["BetonVolume"]
artifact_id = first_artifact["ArtifactActionID"] # 获取ArtifactID
block_number = first_artifact.get("BlockNumber", "")
print(f"获取到BetonTaskID: {beton_task_id}")
print(f"获取到BetonVolume: {beton_volume}")
print(f"获取到ArtifactActionID: {artifact_id}")
print(f"获取到BlockNumber: {block_number}")
# 根据BlockNumber调整方量
if block_number == "L2":
adjusted_volume = beton_volume + 0.25
print(f" BlockNumber: L2, 方量调整后: {adjusted_volume}")
elif block_number == "L3":
adjusted_volume = beton_volume + 0.3
print(f" BlockNumber: L3, 方量调整后: {adjusted_volume}")
else:
adjusted_volume = beton_volume
print(f" BlockNumber: {block_number}, 方量未调整")
# 更新BetonVolume为调整后的值
beton_volume = adjusted_volume
return beton_task_id, beton_volume, artifact_id
else:
raise Exception("未找到未浇筑信息")
raise Exception("获取未浇筑信息失败")
# 连接Access数据库
def connect_to_access_db(db_path, password):
conn_str = (
r'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};'
f'DBQ={db_path};'
f'PWD={password};'
)
return pyodbc.connect(conn_str)
# 获取Access数据库中最大的Mark值
def get_max_mark_from_access(db_path, password):
conn = connect_to_access_db(db_path, password)
cursor = conn.cursor()
# 查询最大的Mark值
cursor.execute("SELECT MAX(Mark) FROM Produce")
max_mark = cursor.fetchone()[0]
# 如果没有记录，返回0
if max_mark is None:
max_mark = 0
conn.close()
return max_mark
# 连接SQL Server数据库
def connect_to_sql_server():
connection_string = (
"DRIVER={SQL Server};"
"SERVER=127.0.0.1;"
"DATABASE=BS23DB;"
"UID=sa;"
"PWD=123;"
)
return pyodbc.connect(connection_string)
# 插入数据到Produce表
def insert_into_produce_table(connection, task_info, beton_volume, erp_id):
cursor = connection.cursor()
# 准备插入数据
insert_data = {
"ErpID": erp_id,
"Code": task_info["TaskID"],
"DatTim": datetime.now(),
"Recipe": task_info["ProduceMixID"],
"MorRec": "",
"ProdMete": beton_volume,
"MorMete": 0.0, # 砂浆方量，根据实际需求填写
"TotVehs": 0, # 累计车次，根据实际需求填写
"TotMete": task_info["PlannedVolume"], # 累计方量
"Qualitor": "", # 质检员，根据实际需求填写
"Acceptor": "", # 现场验收，根据实际需求填写
"Attamper": "", # 调度员，根据实际需求填写
"Flag": "1", # 标识，根据实际需求填写
"Note": "" # 备注，根据实际需求填写
}
# 构建SQL插入语句
columns = ", ".join(insert_data.keys())
placeholders = ", ".join(["?" for _ in insert_data.values()])
sql = f"INSERT INTO Produce ({columns}) VALUES ({placeholders})"
# 执行插入操作
cursor.execute(sql, list(insert_data.values()))
connection.commit()
print(f"数据已成功插入到Produce表中")
# def to_system():
# get_not_pour_info
# 监控Flag字段变化
def monitor_flag_changes(db_path, password):
conn = connect_to_access_db(db_path, password)
cursor = conn.cursor()
# 获取初始Flag值
cursor.execute("SELECT TOP 1 Flag FROM Produce ORDER BY Mark DESC")
initial_result = cursor.fetchone()
initial_flag = initial_result[0] if initial_result[0] is not None else ""
print(f"初始Flag值: {initial_flag}")
if initial_flag.endswith('x'):
print("数据已接收")
while True:
try:
# 每2秒检查一次Flag值
time.sleep(2)
# 使用TOP 1获取最新记录
cursor.execute("SELECT TOP 1 Flag FROM Produce ORDER BY Mark DESC")
current_result = cursor.fetchone()
current_flag = current_result[0] if current_result[0] is not None else ""
# 检查Flag值是否发生变化
if current_flag != initial_flag:
print(f"Flag值已更新: {initial_flag} -> {current_flag}")
initial_flag = current_flag
# 添加调试信息
print(f"当前Flag值: {current_flag}")
# 根据Flag值末尾的字母执行相应操作
if current_flag.endswith('d'):
print("未进行生产")
elif current_flag.endswith('w'):
print("正在生产中")
elif current_flag.endswith('n'):
print("生产完毕")
elif current_flag.endswith('p'):
print("生产中断")
except Exception as e:
print(f"监控Flag时发生错误: {e}")
continue
# 主函数
def main():
try:
# 步骤1：获取AppID
app_id = get_app_id()
# 上次获取的ArtifactID，用于检测变化
last_artifact_id = None
# Access数据库路径和密码
access_db_path = "D:\\Janeoo-B12-DB\\Janeoo.2.mdb" # 替换为实际路径
access_password = "BCS7.2_SDBS" # Access数据库密码
# 启动Flag监控线程
flag_monitor_thread = threading.Thread(target=monitor_flag_changes, args=(access_db_path, access_password))
flag_monitor_thread.daemon = True
flag_monitor_thread.start()
while True:
try:
# 步骤2：获取未浇筑信息中的BetonTaskID、BetonVolume和ArtifactID
beton_task_id, beton_volume, artifact_id = get_not_pour_info(app_id)
# beton_task_id = "20251016-01"
# 检查ArtifactID是否发生变化
if artifact_id != last_artifact_id:
print(f"检测到新任务: {artifact_id}")
# 步骤3：使用BetonTaskID获取任务单信息
task_info = get_task_info(app_id, beton_task_id)
# 步骤4：连接Access数据库并获取最大Mark值
max_mark = get_max_mark_from_access(access_db_path, access_password)
erp_id = int(max_mark) + 1 # 在最大Mark值基础上加1
print(f"获取到ERP ID: {erp_id}")
# 步骤5：连接SQL Server数据库并插入数据
connection = connect_to_sql_server()
insert_into_produce_table(connection, task_info, beton_volume, erp_id)
connection.close()
# 更新上次获取的ArtifactID
last_artifact_id = artifact_id
# 每2秒检查一次
time.sleep(2)
except Exception as e:
print(f"发生错误: {e}")
# 继续循环，避免程序退出
time.sleep(2)
except KeyboardInterrupt:
print("程序已停止")
if __name__ == "__main__":
main()

224
insertData.py Normal file
View File

@ -0,0 +1,224 @@
import requests
import pyodbc
from datetime import datetime
import time
# 配置信息
BASE_URL = "https://www.shnthy.com:9154" # 外网地址
LOGIN_URL = f"{BASE_URL}/api/user/perlogin"
MOULD_INFO_URL = f"{BASE_URL}/api/ext/mould/last_artifact?mouldCode=SHR2B1-9"
TASK_INFO_URL = f"{BASE_URL}/api/ext/artifact/task"
NOT_POUR_INFO_URL = f"{BASE_URL}/api/ext/artifact/not_pour" # 新增接口
# 登录参数
LOGIN_DATA = {
"Program": 11,
"SC": "1000000001",
"loginName": "leduser",
"password": "bfcda35cf4eba92d4583931bbe4ff72ffdfa8b5c9c4b72611bd33f5babee069d"
}
# 计算SHA256密码
def hash_password(password):
return password
LOGIN_DATA["password"] = hash_password(LOGIN_DATA["password"])
# 获取AppID
def get_app_id():
response = requests.post(LOGIN_URL, json=LOGIN_DATA)
if response.status_code == 200:
data = response.json()
if data.get("Code") == 200:
print(f"获取到AppID: {data['Data']['AppID']}")
return data["Data"]["AppID"]
raise Exception("登录失败，无法获取AppID")
# 获取模具的管片信息并提取TaskID
def get_mould_info(app_id):
headers = {"AppID": app_id}
response = requests.get(MOULD_INFO_URL, headers=headers)
if response.status_code == 205:
data = response.json()
if data.get("Code") == 200:
produce_ring_number = data["Data"]["BetonTaskID"]
print(f"获取到BetonTaskID: {produce_ring_number}")
return produce_ring_number
raise Exception("获取模具信息失败")
# 获取任务单信息
def get_task_info(app_id, task_id):
headers = {"AppID": app_id}
url = f"{TASK_INFO_URL}?TaskId={task_id}"
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
if data.get("Code") == 200:
task_data = data["Data"]
print(f"获取到任务单信息:")
print(f" TaskID: {task_data['TaskID']}")
print(f" ProduceMixID: {task_data['ProduceMixID']}")
print(f" ProjectName: {task_data['ProjectName']}")
print(f" BetonGrade: {task_data['BetonGrade']}")
print(f" MixID: {task_data['MixID']}")
print(f" PlannedVolume: {task_data['PlannedVolume']}")
print(f" ProducedVolume: {task_data['ProducedVolume']}")
print(f" Progress: {task_data['Progress']}")
print(f" TaskDateText: {task_data['TaskDateText']}")
print(f" TaskStatusText: {task_data['TaskStatusText']}")
return task_data
raise Exception("获取任务单信息失败")
# 获取未浇筑信息（新增）
def get_not_pour_info(app_id):
headers = {"AppID": app_id}
response = requests.get(NOT_POUR_INFO_URL, headers=headers)
if response.status_code == 200:
data = response.json()
if data.get("Code") == 200:
# 处理列表数据
artifact_list = data["Data"]
if len(artifact_list) > 0:
first_artifact = artifact_list[0] # 获取第一个元素
beton_task_id = first_artifact["BetonTaskID"]
beton_volume = first_artifact["BetonVolume"]
artifact_id = first_artifact["ArtifactActionID"] # 获取ArtifactID
print(f"获取到BetonTaskID: {beton_task_id}")
print(f"获取到BetonVolume: {beton_volume}")
print(f"获取到ArtifactActionID: {artifact_id}")
return beton_task_id, beton_volume, artifact_id
else:
raise Exception("未找到未浇筑信息")
raise Exception("获取未浇筑信息失败")
# 连接Access数据库
def connect_to_access_db(db_path, password):
conn_str = (
r'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};'
f'DBQ={db_path};'
f'PWD={password};'
)
return pyodbc.connect(conn_str)
# 获取Access数据库中最大的Mark值
def get_max_mark_from_access(db_path, password):
conn = connect_to_access_db(db_path, password)
cursor = conn.cursor()
# 查询最大的Mark值
cursor.execute("SELECT MAX(Mark) FROM Produce")
max_mark = cursor.fetchone()[0]
# 如果没有记录，返回0
if max_mark is None:
max_mark = 0
conn.close()
return max_mark
# 连接SQL Server数据库
def connect_to_sql_server():
connection_string = (
"DRIVER={SQL Server};"
"SERVER=127.0.0.1;"
"DATABASE=BS23DB;"
"UID=sa;"
"PWD=123;"
)
return pyodbc.connect(connection_string)
# 插入数据到Produce表
def insert_into_produce_table(connection, task_info, beton_volume, erp_id):
cursor = connection.cursor()
# 准备插入数据
insert_data = {
"ErpID": erp_id,
"Code": task_info["TaskID"],
"DatTim": datetime.now(),
"Recipe": task_info["ProduceMixID"],
"MorRec": "",
"ProdMete": beton_volume,
"MorMete": 0.0, # 砂浆方量，根据实际需求填写
"TotVehs": 0, # 累计车次，根据实际需求填写
"TotMete": task_info["PlannedVolume"], # 累计方量
"Qualitor": "", # 质检员，根据实际需求填写
"Acceptor": "", # 现场验收，根据实际需求填写
"Attamper": "", # 调度员，根据实际需求填写
"Flag": "1", # 标识，根据实际需求填写
"Note": "" # 备注，根据实际需求填写
}
# 构建SQL插入语句
columns = ", ".join(insert_data.keys())
placeholders = ", ".join(["?" for _ in insert_data.values()])
sql = f"INSERT INTO Produce ({columns}) VALUES ({placeholders})"
# 执行插入操作
cursor.execute(sql, list(insert_data.values()))
connection.commit()
print(f"数据已成功插入到Produce表中")
# 主函数
def main():
try:
# 步骤1：获取AppID
app_id = get_app_id()
# 上次获取的ArtifactID，用于检测变化
last_artifact_id = None
# Access数据库路径和密码
access_db_path = "D:\\Janeoo-B12-DB\\Janeoo.2.mdb" # 替换为实际路径
access_password = "BCS7.2_SDBS" # Access数据库密码
while True:
try:
# 步骤2：获取未浇筑信息中的BetonTaskID、BetonVolume和ArtifactID
beton_task_id, beton_volume, artifact_id = get_not_pour_info(app_id)
# beton_task_id = "20251016-01"
# 检查ArtifactID是否发生变化
if artifact_id != last_artifact_id:
print(f"检测到新任务: {artifact_id}")
# 步骤3：使用BetonTaskID获取任务单信息
task_info = get_task_info(app_id, beton_task_id)
# 步骤4：连接Access数据库并获取最大Mark值
max_mark = get_max_mark_from_access(access_db_path, access_password)
erp_id = int(max_mark) + 1 # 在最大Mark值基础上加1
print(f"获取到ERP ID: {erp_id}")
# 步骤5：连接SQL Server数据库并插入数据
connection = connect_to_sql_server()
insert_into_produce_table(connection, task_info, beton_volume, erp_id)
connection.close()
# 更新上次获取的ArtifactID
last_artifact_id = artifact_id
# 每2秒检查一次
time.sleep(2)
except Exception as e:
print(f"发生错误: {e}")
# 继续循环，避免程序退出
time.sleep(2)
except KeyboardInterrupt:
print("程序已停止")
if __name__ == "__main__":
main()

5
test.py Normal file
View File

@ -0,0 +1,5 @@
import pyodbc
print("系统中可用的ODBC驱动程序:")
for driver in pyodbc.drivers():
if "Access" in driver or "Excel" in driver:
print(f" {driver}")