import argparse import os from datetime import datetime import pymysql ARCHIVE_BATCH_ID = "DL_LEGACY_COMPLIANCE_REPAIR_20260624_V1" DEFAULT_SOURCE_ID = "SOURCE_LEGACY_EXTERNAL_INFORMATION_CASEBOOK" def connect(args): password = args.password if password is None: password = os.environ.get("DARKLINE_MYSQL_PASSWORD", "") return pymysql.connect( host=args.host, port=args.port, user=args.user, password=password, database=args.database, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=False, ) def fetch_one(cur, sql, params): cur.execute(sql, params) return cur.fetchone() def upsert(cur, sql, params): cur.execute(sql, params) def main(): parser = argparse.ArgumentParser(description="Repair legacy darkline case structural compliance without fabricating facts.") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=3306) parser.add_argument("--user", default="root") parser.add_argument("--password", default=None) parser.add_argument("--database", default="tianxia") args = parser.parse_args() now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") conn = connect(args) repaired = { "event_gap_rows": 0, "event_gap_links": 0, "manifestation_gap_rows": 0, "alternative_gap_rows": 0, "chain_state_rows": 0, } with conn.cursor() as cur: cur.execute( """ SELECT c.case_id, c.case_title, c.case_level, h.darkline_hypothesis_id FROM dl_case_record c JOIN dl_darkline_hypothesis h ON h.case_id = c.case_id WHERE c.import_batch_id LIKE 'DL_LEGACY_CASE_IMPORT%' ORDER BY c.case_order, c.case_id """ ) cases = cur.fetchall() for case in cases: case_id = case["case_id"] hypothesis_id = case["darkline_hypothesis_id"] case_title = case["case_title"] event_level = case["case_level"] or "UNKNOWN" first_step = fetch_one( cur, "SELECT reasoning_step_id FROM dl_case_reasoning_step WHERE case_id=%s ORDER BY step_order LIMIT 1", (case_id,), ) first_evidence = fetch_one( cur, "SELECT evidence_id FROM dl_evidence WHERE case_id=%s ORDER BY evidence_id LIMIT 1", (case_id,), ) first_expected = fetch_one( cur, "SELECT expected_line_id FROM dl_expected_line WHERE case_id=%s ORDER BY expected_line_id LIMIT 1", (case_id,), ) first_target = fetch_one( cur, "SELECT impact_target_id FROM dl_impact_target WHERE case_id=%s ORDER BY impact_target_id LIMIT 1", (case_id,), ) step_id = first_step["reasoning_step_id"] if first_step else None evidence_id = first_evidence["evidence_id"] if first_evidence else None expected_id = first_expected["expected_line_id"] if first_expected else None target_id = first_target["impact_target_id"] if first_target else None missing_flags = [] event_count = fetch_one(cur, "SELECT COUNT(*) cnt FROM dl_event_node WHERE case_id=%s", (case_id,))["cnt"] if event_count == 0: event_id = f"{case_id}_EVENT_NODE_GAP_REVIEW" upsert( cur, """ INSERT INTO dl_event_node ( event_node_id, darkline_hypothesis_id, case_id, event_title, event_date, available_time, actor_list, target_list, event_type, event_level, source_id, node_status, source_reasoning_step_id, archive_batch_id ) VALUES (%s,%s,%s,%s,NULL,NULL,NULL,NULL,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE node_status=VALUES(node_status), archive_batch_id=VALUES(archive_batch_id) """, ( event_id, hypothesis_id, case_id, f"历史导入缺口：{case_title} 尚未拆解现实事件节点", "EVENT_NODE_ARCHIVE_GAP", event_level, DEFAULT_SOURCE_ID, "HELD_BY_EVENT_NODE_ARCHIVE_GAP", step_id, ARCHIVE_BATCH_ID, ), ) repaired["event_gap_rows"] += 1 missing_flags.append("event_node_gap") if evidence_id: link_id = f"{case_id}_LINK_EVENT_GAP_REVIEW" upsert( cur, """ INSERT INTO dl_evidence_node_link ( evidence_node_link_id, evidence_id, event_node_id, darkline_hypothesis_id, expected_line_id, link_role, archive_batch_id ) VALUES (%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE event_node_id=VALUES(event_node_id), expected_line_id=VALUES(expected_line_id), link_role=VALUES(link_role), archive_batch_id=VALUES(archive_batch_id) """, ( link_id, evidence_id, event_id, hypothesis_id, expected_id, "GAP_REVIEW_EVIDENCE", ARCHIVE_BATCH_ID, ), ) repaired["event_gap_links"] += 1 manifestation_count = fetch_one(cur, "SELECT COUNT(*) cnt FROM dl_manifestation_bridge WHERE case_id=%s", (case_id,))["cnt"] bridge_id = None if manifestation_count == 0: bridge_id = f"{case_id}_MANIFESTATION_GAP_REVIEW" upsert( cur, """ INSERT INTO dl_manifestation_bridge ( bridge_id, darkline_hypothesis_id, case_id, expected_line_id, impact_target_id, output_layer, manifestation_time, manifestation_value, control_group_id, support_status, source_reasoning_step_id, archive_batch_id ) VALUES (%s,%s,%s,%s,%s,%s,NULL,%s,NULL,%s,%s,%s) ON DUPLICATE KEY UPDATE manifestation_value=VALUES(manifestation_value), support_status=VALUES(support_status), archive_batch_id=VALUES(archive_batch_id) """, ( bridge_id, hypothesis_id, case_id, expected_id, target_id, "MARKET_MANIFESTATION_REVIEW", "历史导入缺口：尚未归档市场或输出层显影；不得视为已验证。", "HELD_BY_MARKET_MANIFESTATION_GAP", step_id, ARCHIVE_BATCH_ID, ), ) repaired["manifestation_gap_rows"] += 1 missing_flags.append("manifestation_gap") else: first_bridge = fetch_one( cur, "SELECT bridge_id FROM dl_manifestation_bridge WHERE case_id=%s ORDER BY bridge_id LIMIT 1", (case_id,), ) bridge_id = first_bridge["bridge_id"] if first_bridge else None alternative_count = fetch_one(cur, "SELECT COUNT(*) cnt FROM dl_alternative_explanation WHERE case_id=%s", (case_id,))["cnt"] if alternative_count == 0: alternative_id = f"{case_id}_ALTERNATIVE_GAP_REVIEW" upsert( cur, """ INSERT INTO dl_alternative_explanation ( alternative_id, darkline_hypothesis_id, case_id, bridge_id, alternative_type, explanation, strength, current_status, source_reasoning_step_id, archive_batch_id ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE bridge_id=VALUES(bridge_id), explanation=VALUES(explanation), current_status=VALUES(current_status), archive_batch_id=VALUES(archive_batch_id) """, ( alternative_id, hypothesis_id, case_id, bridge_id, "ALTERNATIVE_EXPLANATION_REVIEW_PENDING", "历史导入缺口：替代解释未完整归档。复用前必须检查市场普涨、行业风、单股动能、其他暗线和数据缺口。", "REVIEW", "HELD_BY_ALTERNATIVE_EXPLANATION_GAP", step_id, ARCHIVE_BATCH_ID, ), ) repaired["alternative_gap_rows"] += 1 missing_flags.append("alternative_gap") note = "Legacy case structural compliance audited." if missing_flags: note += " Gap placeholders added: " + ",".join(missing_flags) + ". These rows are review markers, not confirmed facts." chain_status = "LEGACY_STRUCTURAL_COMPLIANCE_REPAIRED" reality_status = "REVIEW_GAP_EXPLICITLY_MARKED" market_status = "REVIEW_GAP_EXPLICITLY_MARKED" readout = "CASEBOOK_WITH_EXPLICIT_GAPS" else: note += " Full chain rows were already present before repair." chain_status = "LEGACY_FULL_CHAIN_PRESENT" reality_status = "LEGACY_FULL_CHAIN_PRESENT" market_status = "LEGACY_FULL_CHAIN_PRESENT" readout = "CASEBOOK_FULL_CHAIN_IMPORTED" state_id = f"{case_id}_STATE_LEGACY_COMPLIANCE_V1" upsert( cur, """ INSERT INTO dl_chain_state ( state_record_id, darkline_hypothesis_id, case_id, chain_clarity_status, reality_confirmation_status, market_manifestation_status, validation_readout_status, current_consumption_level, state_record_time, note ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE chain_clarity_status=VALUES(chain_clarity_status), reality_confirmation_status=VALUES(reality_confirmation_status), market_manifestation_status=VALUES(market_manifestation_status), validation_readout_status=VALUES(validation_readout_status), current_consumption_level=VALUES(current_consumption_level), state_record_time=VALUES(state_record_time), note=VALUES(note) """, ( state_id, hypothesis_id, case_id, chain_status, reality_status, market_status, readout, "CASEBOOK", now, note, ), ) repaired["chain_state_rows"] += 1 conn.commit() print({"case_count": len(cases), **repaired}) if __name__ == "__main__": main()