462 lines
18 KiB
Python
462 lines
18 KiB
Python
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"
|
||
}
|
||
|
||
# 全局变量用于存储需要监控的任务ERP ID
|
||
monitored_tasks = set() # 存储需要监控的erp_id
|
||
inserted_tasks = {} # {erp_id: artifact_id}
|
||
tasks_lock = threading.Lock()
|
||
|
||
|
||
# 计算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_all_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"]
|
||
tasks = []
|
||
for artifact in artifact_list:
|
||
beton_task_id = artifact["BetonTaskID"]
|
||
beton_volume = artifact["BetonVolume"]
|
||
artifact_id = artifact["ArtifactActionID"]
|
||
block_number = artifact.get("BlockNumber", "")
|
||
|
||
# 根据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}, 方量未调整")
|
||
|
||
tasks.append({
|
||
"beton_task_id": beton_task_id,
|
||
"beton_volume": adjusted_volume,
|
||
"artifact_id": artifact_id,
|
||
"block_number": block_number
|
||
})
|
||
|
||
return tasks
|
||
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表
|
||
# 在 insert_into_produce_table 函数中添加调用同事的保存函数
|
||
def insert_into_produce_table(connection, task_info, beton_volume, erp_id, artifact_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表中，ERP ID: {erp_id}")
|
||
|
||
# 记录任务映射关系
|
||
with tasks_lock:
|
||
inserted_tasks[erp_id] = artifact_id
|
||
monitored_tasks.add(erp_id)
|
||
print(f"任务 {erp_id} (ArtifactID: {artifact_id}) 已添加到监控列表")
|
||
|
||
# 调用同事提供的保存函数，将数据保存到自定义数据表
|
||
try:
|
||
# 假设同事提供的函数名为 save_to_custom_table
|
||
# 参数包括: MISID(即erp_id), Flag, TaskID, ProduceMixID, ProjectName, BetonGrade, 调整后的方量
|
||
save_to_custom_table(
|
||
misid=erp_id,
|
||
flag="1", # 初始Flag值
|
||
task_id=task_info["TaskID"],
|
||
produce_mix_id=task_info["ProduceMixID"],
|
||
project_name=task_info["ProjectName"],
|
||
beton_grade=task_info["BetonGrade"],
|
||
adjusted_volume=beton_volume # 已经调整后的方量
|
||
)
|
||
print(f"任务 {erp_id} 的数据已保存到自定义数据表")
|
||
except Exception as e:
|
||
print(f"调用保存函数时出错: {e}")
|
||
|
||
return erp_id
|
||
|
||
# 监控Access数据库中特定任务的Flag字段变化
|
||
# 监控Access数据库中特定任务的Flag字段变化
|
||
def monitor_access_flag_changes(access_db_path, access_password):
|
||
"""监控Access数据库中派发任务的Flag状态"""
|
||
|
||
# 存储任务的当前状态
|
||
task_flags = {}
|
||
|
||
print("开始监控Access数据库中派发任务的Flag状态")
|
||
|
||
while True:
|
||
try:
|
||
# 每2秒检查一次
|
||
time.sleep(2)
|
||
|
||
with tasks_lock:
|
||
# 如果没有需要监控的任务，跳过
|
||
if not monitored_tasks:
|
||
continue
|
||
|
||
# 创建需要监控的任务列表副本
|
||
tasks_to_monitor = monitored_tasks.copy()
|
||
|
||
# 首先检查SQL Server中任务是否还存在
|
||
sql_conn = connect_to_sql_server()
|
||
sql_cursor = sql_conn.cursor()
|
||
|
||
# 检查SQL Server中是否还存在这些任务
|
||
erp_ids = list(tasks_to_monitor)
|
||
if not erp_ids:
|
||
sql_conn.close()
|
||
continue
|
||
|
||
erp_ids_str = [str(erp_id) for erp_id in erp_ids]
|
||
placeholders = ','.join('?' * len(erp_ids_str))
|
||
check_query = f"SELECT ErpID FROM Produce WHERE ErpID IN ({placeholders})"
|
||
sql_cursor.execute(check_query, erp_ids_str)
|
||
sql_results = sql_cursor.fetchall()
|
||
sql_conn.close()
|
||
|
||
# 分离已删除和未删除的任务
|
||
existing_tasks_in_sql = {str(row[0]) for row in sql_results}
|
||
deleted_from_sql_tasks = set(erp_ids_str) - existing_tasks_in_sql
|
||
|
||
print(f"SQL Server中仍存在的任务: {existing_tasks_in_sql}")
|
||
print(f"已从SQL Server删除的任务: {deleted_from_sql_tasks}")
|
||
|
||
# 只有已从SQL Server删除的任务才需要在Access中查找
|
||
if not deleted_from_sql_tasks:
|
||
continue
|
||
|
||
# 连接Access数据库
|
||
conn = connect_to_access_db(access_db_path, access_password)
|
||
cursor = conn.cursor()
|
||
|
||
# 查询Access数据库中已删除任务的状态
|
||
placeholders = ','.join('?' * len(list(deleted_from_sql_tasks)))
|
||
query = f"SELECT MISID, Flag FROM Produce WHERE MISID IN ({placeholders})"
|
||
|
||
# 添加调试信息
|
||
print(f"执行查询: {query}")
|
||
print(f"查询参数: {list(deleted_from_sql_tasks)}")
|
||
|
||
cursor.execute(query, list(deleted_from_sql_tasks))
|
||
results = cursor.fetchall()
|
||
|
||
# 添加调试信息
|
||
print(f"查询返回结果数量: {len(results)}")
|
||
|
||
# 如果没有查询到结果，检查数据库中的实际数据
|
||
if len(results) == 0 and deleted_from_sql_tasks:
|
||
print("未找到匹配记录，检查数据库中的实际数据:")
|
||
try:
|
||
cursor.execute("SELECT TOP 5 Mark, Flag FROM Produce ORDER BY Mark DESC")
|
||
sample_data = cursor.fetchall()
|
||
print(f"数据库中最近的5条记录: {sample_data}")
|
||
except Exception as sample_e:
|
||
print(f"查询样本数据时出错: {sample_e}")
|
||
|
||
# 处理查询结果
|
||
current_tasks = {}
|
||
for row in results:
|
||
mark = row[0]
|
||
flag = row[1] if row[1] is not None else ""
|
||
current_tasks[mark] = flag
|
||
print(f"查询到记录 - MISID: {mark}, Flag: '{flag}'") # 调试信息
|
||
|
||
# 检查每个已删除任务的状态变化
|
||
for erp_id_str in deleted_from_sql_tasks:
|
||
erp_id = int(erp_id_str)
|
||
current_flag = current_tasks.get(erp_id_str, "")
|
||
previous_flag = task_flags.get(erp_id_str, "")
|
||
|
||
# 添加调试信息
|
||
print(f"检查任务 {erp_id} - 当前Flag: '{current_flag}', 之前Flag: '{previous_flag}'")
|
||
|
||
# 如果状态发生变化
|
||
if current_flag != previous_flag:
|
||
with tasks_lock:
|
||
artifact_id = inserted_tasks.get(erp_id, "Unknown")
|
||
print(
|
||
f"派发任务 ErpID {erp_id} (ArtifactID: {artifact_id}) 的Flag值已更新: {previous_flag} -> {current_flag}")
|
||
task_flags[erp_id_str] = current_flag
|
||
|
||
# 根据Flag值末尾的字母执行相应操作并更新自定义数据表状态
|
||
if current_flag.endswith('d'):
|
||
print(f"派发任务 ErpID {erp_id}: 未进行生产")
|
||
# 调用同事提供的状态更新函数
|
||
try:
|
||
update_custom_table_status(erp_id, "未进行生产")
|
||
except Exception as e:
|
||
print(f"更新状态时出错: {e}")
|
||
elif current_flag.endswith('w'):
|
||
print(f"派发任务 ErpID {erp_id}: 正在生产中")
|
||
# 调用同事提供的状态更新函数
|
||
try:
|
||
update_custom_table_status(erp_id, "正在生产中")
|
||
except Exception as e:
|
||
print(f"更新状态时出错: {e}")
|
||
elif current_flag.endswith('n'):
|
||
print(f"派发任务 ErpID {erp_id}: 生产完毕")
|
||
# 任务完成，可以从监控列表中移除
|
||
with tasks_lock:
|
||
monitored_tasks.discard(erp_id)
|
||
print(f"派发任务 ErpID {erp_id} 已完成，停止监控")
|
||
# 调用同事提供的状态更新函数
|
||
try:
|
||
update_custom_table_status(erp_id, "生产完毕")
|
||
except Exception as e:
|
||
print(f"更新状态时出错: {e}")
|
||
elif current_flag.endswith('p'):
|
||
print(f"派发任务 ErpID {erp_id}: 生产中断")
|
||
# 任务中断，可以从监控列表中移除
|
||
with tasks_lock:
|
||
monitored_tasks.discard(erp_id)
|
||
print(f"派发任务 ErpID {erp_id} 已中断，停止监控")
|
||
# 调用同事提供的状态更新函数
|
||
try:
|
||
update_custom_table_status(erp_id, "生产中断")
|
||
except Exception as e:
|
||
print(f"更新状态时出错: {e}")
|
||
elif current_flag.endswith('x'):
|
||
print(f"派发任务 ErpID {erp_id}: 数据已接收")
|
||
# 调用同事提供的状态更新函数
|
||
try:
|
||
update_custom_table_status(erp_id, "数据已接收")
|
||
except Exception as e:
|
||
print(f"更新状态时出错: {e}")
|
||
|
||
# 检查是否有任务记录已被删除（不在查询结果中但仍在监控列表中）
|
||
# 这表示任务可能已完成或从系统中移除
|
||
missing_tasks = set(deleted_from_sql_tasks) - set(current_tasks.keys())
|
||
if missing_tasks:
|
||
for erp_id_str in missing_tasks:
|
||
erp_id = int(erp_id_str)
|
||
with tasks_lock:
|
||
artifact_id = inserted_tasks.get(erp_id, "Unknown")
|
||
monitored_tasks.discard(erp_id)
|
||
inserted_tasks.pop(erp_id, None)
|
||
task_flags.pop(erp_id_str, None)
|
||
print(f"派发任务 ErpID {erp_id} (ArtifactID: {artifact_id}) 记录已从Access数据库中删除或完成")
|
||
|
||
conn.close()
|
||
|
||
except Exception as e:
|
||
print(f"监控Access数据库 Flag时发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
continue
|
||
|
||
|
||
|
||
# 在 main 函数中修改任务处理逻辑
|
||
def main():
|
||
try:
|
||
# 步骤1：获取AppID
|
||
app_id = get_app_id()
|
||
|
||
# 存储上次获取的所有ArtifactID
|
||
last_artifact_ids = set()
|
||
|
||
# Access数据库路径和密码
|
||
access_db_path = "D:\\Janeoo-B12-DB\\Janeoo.2.mdb" # 替换为实际路径
|
||
access_password = "BCS7.2_SDBS" # Access数据库密码
|
||
|
||
# 启动Access数据库Flag监控线程
|
||
access_monitor_thread = threading.Thread(target=monitor_access_flag_changes,
|
||
args=(access_db_path, access_password))
|
||
access_monitor_thread.daemon = True
|
||
access_monitor_thread.start()
|
||
|
||
while True:
|
||
try:
|
||
# 步骤2：获取所有未浇筑信息
|
||
tasks = get_all_not_pour_info(app_id)
|
||
current_artifact_ids = {task["artifact_id"] for task in tasks}
|
||
|
||
# 检查是否有新任务
|
||
new_artifact_ids = current_artifact_ids - last_artifact_ids
|
||
if new_artifact_ids:
|
||
print(f"检测到 {len(new_artifact_ids)} 个新任务")
|
||
|
||
for task in tasks:
|
||
if task["artifact_id"] in new_artifact_ids:
|
||
# 检查 block_number 是否为 "F"
|
||
if task["block_number"] == "F":
|
||
print(f"任务 {task['artifact_id']} 的 block_number 为 'F'，跳过派单")
|
||
continue
|
||
|
||
print(f"处理新任务: {task['artifact_id']}")
|
||
|
||
# 步骤3：获取任务单信息
|
||
task_info = get_task_info(app_id, task["beton_task_id"])
|
||
|
||
# 步骤4：连接Access数据库并获取最大Mark值
|
||
max_mark = get_max_mark_from_access(access_db_path, access_password)
|
||
erp_id = int(max_mark) + 1
|
||
print(f"获取到ERP ID: {erp_id}")
|
||
|
||
# 步骤5：连接SQL Server数据库并插入数据
|
||
connection = connect_to_sql_server()
|
||
insert_into_produce_table(connection, task_info, task["beton_volume"], erp_id,
|
||
task["artifact_id"])
|
||
connection.close()
|
||
|
||
# 更新上次获取的ArtifactID集合
|
||
last_artifact_ids = current_artifact_ids
|
||
|
||
# 每2秒检查一次
|
||
time.sleep(2)
|
||
|
||
except Exception as e:
|
||
print(f"发生错误: {e}")
|
||
# 继续循环，避免程序退出
|
||
time.sleep(2)
|
||
|
||
except KeyboardInterrupt:
|
||
print("程序已停止")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|