From 09f84c4ecae2a71e07f6dd64e6066563599a62b6 Mon Sep 17 00:00:00 2001
From: Administrator <admin@example.com>
Date: 星期二, 17 六月 2025 17:59:40 +0800
Subject: [PATCH] ‘功能完善
---
server/local_data_server.py | 168 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 168 insertions(+), 0 deletions(-)
diff --git a/server/local_data_server.py b/server/local_data_server.py
new file mode 100644
index 0000000..2bad72d
--- /dev/null
+++ b/server/local_data_server.py
@@ -0,0 +1,168 @@
+"""
+鏈湴http鏈嶅姟鍣�
+"""
+import json
+from http.server import SimpleHTTPRequestHandler, HTTPServer
+import urllib.parse as urlparse
+from urllib.parse import parse_qs
+
+import constant
+from db.mysql_data_delegate import Mysqldb
+
+from strategy.low_suction_strategy import LowSuctionOriginDataExportManager
+from strategy.strategy_variable_factory import DataLoader
+from third_data import kpl_util
+from utils import tool, output_util, huaxin_util
+
+
+class CORSRequestHandler(SimpleHTTPRequestHandler):
+ def end_headers(self):
+ # 娣诲姞 CORS 澶�
+ self.send_header('Access-Control-Allow-Origin', '*')
+ self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
+ self.send_header('Access-Control-Allow-Headers', 'Content-Type')
+ super().end_headers()
+
+ # 澶勭悊 OPTIONS 棰勬璇锋眰
+ def do_OPTIONS(self):
+ self.send_response(200)
+ self.end_headers()
+
+ def do_GET(self) -> None:
+ path = self.path
+ url = urlparse.urlparse(path)
+ ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
+ response_data = ""
+ if url.path == "/get_limit_up_plate_list":
+ date = ps_dict.get("date")
+ time_str = ps_dict.get("time")
+ __LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(date)
+ # [(浠ｇ爜, 浠ｇ爜鍚嶇О, 娑ㄥ仠鍘熷洜, 娑ㄥ仠鏃堕棿, 楂樺害淇℃伅, 鑷敱娴侀�氬競鍊硷紝鏄惁鐐告澘)]
+ results = __LowSuctionOriginDataExportManager.export_current_limit_up_records()
+ results = [x for x in results if tool.to_time_str(int(x[3])) <= time_str]
+ code_plates = __LowSuctionOriginDataExportManager.export_code_plates()
+ # 鏈�缁堟定鍋滅殑浠ｇ爜
+ limit_up_codes = [x[0] for x in results if not x[6]]
+ if date == tool.get_now_date_str():
+ limit_up_list = __LowSuctionOriginDataExportManager.export_limit_up_list()
+ if limit_up_list:
+ limit_up_list = [x for x in limit_up_list if x[0][:8] <= time_str]
+ limit_up_list = limit_up_list[-1][1]
+ limit_up_codes = [x[0] for x in limit_up_list]
+
+ code_infos = {x[0]: x for x in results}
+ plate_codes = {}
+ for code in code_infos.keys():
+ plates = code_plates.get(code)
+ if not plates:
+ plates = {kpl_util.filter_block(code_infos[code][2])}
+ plates -= constant.KPL_INVALID_BLOCKS
+ for p in plates:
+ if p not in plate_codes:
+ plate_codes[p] = set()
+ plate_codes[p].add(code)
+ # (鏉垮潡鍚嶇О, 娑ㄥ仠鏁帮紝鐐告澘鏁�)
+ data = [(p, len(plate_codes[p]), len([code for code in plate_codes[p] if code not in limit_up_codes])) for p
+ in
+ plate_codes]
+ data.sort(key=lambda x: x[1] - x[2], reverse=True)
+ response_data = json.dumps({"code": 0, "data": data})
+
+ elif url.path == "/get_plate_codes":
+ date = ps_dict.get("date")
+ plate = ps_dict.get("plate")
+ time_str = ps_dict.get("time")
+ __LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(date)
+ results = __LowSuctionOriginDataExportManager.export_current_limit_up_records()
+ for r in results:
+ r[3] = tool.to_time_str(int(r[3]))
+ r[5] = output_util.money_desc(r[5])
+ results = [x for x in results if x[3] <= time_str]
+ # 鏈�缁堟定鍋滅殑浠ｇ爜
+ limit_up_codes = [x[0] for x in results if not x[6]]
+ if date == tool.get_now_date_str():
+ limit_up_list = __LowSuctionOriginDataExportManager.export_limit_up_list()
+ if limit_up_list:
+ limit_up_list = [x for x in limit_up_list if x[0][:8] <= time_str]
+ limit_up_list = limit_up_list[-1][1]
+ limit_up_codes = [x[0] for x in limit_up_list]
+ for x in results:
+ x[6] = x[0] not in limit_up_codes
+
+ # [(浠ｇ爜, 浠ｇ爜鍚嶇О, 娑ㄥ仠鍘熷洜, 娑ㄥ仠鏃堕棿, 楂樺害淇℃伅, 鑷敱娴侀�氬競鍊硷紝鏄惁鐐告澘)]
+ code_plates = __LowSuctionOriginDataExportManager.export_code_plates()
+ code_infos = {x[0]: x for x in results}
+ plate_codes = {}
+ for code in code_infos.keys():
+ plates = code_plates.get(code)
+ if not plates:
+ plates = {kpl_util.filter_block(code_infos[code][2])}
+ plates -= constant.KPL_INVALID_BLOCKS
+ for p in plates:
+ if p not in plate_codes:
+ plate_codes[p] = set()
+ plate_codes[p].add(code)
+ codes = plate_codes.get(plate)
+ datas = [code_infos[code] for code in codes]
+ datas.sort(key=lambda x: x[3])
+ response_data = json.dumps({"code": 0, "data": datas})
+ elif url.path == "/get_big_order_list":
+ date = ps_dict.get("date")
+ code = ps_dict.get("code")
+ __LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(date)
+ big_orders_dict = __LowSuctionOriginDataExportManager.export_all_big_order_deal(200e4)
+ datas = big_orders_dict.get(code, [])
+ for x in datas:
+ x[2][3] = huaxin_util.convert_time(x[2][3], with_ms=False)
+ if len(x[2]) > 5:
+ x[2][5] = huaxin_util.convert_time(x[2][5], with_ms=False)
+ order_ids = set()
+ datas.reverse()
+ fdatas = []
+ for d in datas:
+ if d[2][0] in order_ids:
+ continue
+ fdatas.append(d)
+ order_ids.add(d[2][0])
+
+ response_data = json.dumps({"code": 0, "data": fdatas})
+ elif url.path == "/get_block_in_datas":
+ date = ps_dict.get("date")
+ time_str = ps_dict.get("time")
+ __LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(date)
+ block_in_datas = __LowSuctionOriginDataExportManager.export_block_in_datas()
+ fdatas = []
+ for d in reversed(block_in_datas):
+ if d[0] <= time_str:
+ fdatas = d[1]
+ break
+ fdatas = [(x[0], output_util.money_desc(x[1])) for x in fdatas]
+ response_data = json.dumps({"code": 0, "data": fdatas})
+
+ elif url.path == "/get_codes_by_jx_plates":
+ # 鏍规嵁绮鹃�夋澘鍧楄幏鍙栦唬鐮�
+ plates = ps_dict.get("plates")
+ date = ps_dict.get("date")
+ plates = set(json.loads(plates))
+ sql = f" select code, jx_blocks from kpl_code_blocks where day = '{date}' and " + " and ".join(
+ [f"jx_blocks like '%{p}%'" for p in plates])
+ datas = Mysqldb().select_all(sql)
+ fdatas = [(x[0], "銆�".join([ f"<red>{dd}</red>" if dd in plates else dd for dd in json.loads(x[1])])) for x in datas if len(set(json.loads(x[1])) & plates) == len(plates)]
+ response_data = json.dumps({"code": 0, "data": fdatas})
+
+ print("GET璇锋眰")
+ self.send_response(200)
+ # 鍙戠粰璇锋眰瀹㈡埛绔殑鍝嶅簲鏁版嵁
+ self.send_header('Content-type', 'application/json')
+ self.end_headers()
+ self.wfile.write(response_data.encode())
+
+ def do_POST(self) -> None:
+ print("POST璇锋眰")
+
+
+if __name__ == '__main__':
+ server_address = ('', 8000)
+ httpd = HTTPServer(server_address, CORSRequestHandler)
+ print("Server running on http://localhost:8000")
+ httpd.serve_forever()
--
Gitblit v1.8.0