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()
|
