姚章浩
f5e0f8d20f
build / build (push) Successful in 1m7s
将可维护总表与可替换2.0镜像分离，稳定来源关联和原始事件键保留； 源端更新/删除不覆盖主表或人工复核。首次收录保留原评估审计， 主表输入/复核独立持久化；历史原标注不冒充新AI或冻结证据。 历史导入绑定源哈希、逐行批准清单与真实恢复备份；三字段精确去重， 批次及原字段留档，重复执行不覆盖后续修改。仅操作本地治理库。 验证：完整make verify退出0；镜像与区域两组隔离PG race退出0； 真实恢复数据隔离导入12522、跳过792、暂缓1，总表15050； 重放保留模拟修改，结构down拒绝丢历史；真实浏览器第41周551条完整， 历史原标注和独立主表复核读回通过，移动布局及JS异常0。
198 lines
18 KiB
Python
198 lines
18 KiB
Python
"""Hash-bound historical import into the local governance master, never the source DB.
|
|
|
|
Requires an approved row decision ledger and a genuinely restored native backup.
|
|
All employee/answer-bearing subprocess I/O is private; stdout contains counts only.
|
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
|
import collections
|
|
import datetime as dt
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import posixpath
|
|
import re
|
|
import subprocess
|
|
import uuid
|
|
import xml.etree.ElementTree as ET
|
|
import zipfile
|
|
|
|
PG = "knowledge-governance-p1-postgres-1"
|
|
TZ = dt.timezone(dt.timedelta(hours=8))
|
|
NS = {"s": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
|
|
STATUSES = {"new_import_candidate", "existing_2_0_duplicate", "missing_key_review"}
|
|
|
|
def sha(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
def text(value: str) -> str:
|
|
return value.replace("\r\n", "\n").replace("\r", "\n").strip()
|
|
|
|
def literal(value: str) -> str:
|
|
return "'" + value.replace("'", "''") + "'"
|
|
|
|
def read_rows(path: Path, expected_sha: str) -> tuple[dict[str,str],list[tuple[int,dict[str,str]]]]:
|
|
content=path.read_bytes()
|
|
if sha(content)!=expected_sha:
|
|
raise ValueError("workbook changed before snapshot parsing")
|
|
with zipfile.ZipFile(io.BytesIO(content)) as z:
|
|
if z.testzip() is not None:
|
|
raise ValueError("workbook ZIP integrity failed")
|
|
workbook = ET.fromstring(z.read("xl/workbook.xml"))
|
|
sheets = workbook.find("s:sheets", NS)
|
|
if sheets is None or len(sheets)==0:
|
|
raise ValueError("workbook contains no worksheets")
|
|
sheet = sheets[0]
|
|
if sheet.attrib["name"] != "问题流水表":
|
|
raise ValueError("approved first worksheet not found")
|
|
rid = sheet.attrib["{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"]
|
|
relationships = ET.fromstring(z.read("xl/_rels/workbook.xml.rels"))
|
|
target = next(r.attrib["Target"] for r in relationships if r.attrib["Id"] == rid)
|
|
target = target.lstrip("/") if target.startswith("/") else posixpath.normpath("xl/"+target)
|
|
shared = []
|
|
if "xl/sharedStrings.xml" in z.namelist():
|
|
for _, item in ET.iterparse(z.open("xl/sharedStrings.xml"),events=("end",)):
|
|
if item.tag.endswith("}si"):
|
|
shared.append("".join(n.text or "" for n in item.iter() if n.tag.endswith("}t")))
|
|
item.clear()
|
|
def value(cell):
|
|
v=cell.find("s:v",NS)
|
|
if cell.attrib.get("t")=="s" and v is not None:
|
|
return shared[int(v.text)]
|
|
if cell.attrib.get("t")=="inlineStr":
|
|
return "".join(n.text or "" for n in cell.iter() if n.tag.endswith("}t"))
|
|
return (v.text or "") if v is not None else ""
|
|
headers={}; rows=[]
|
|
for _, item in ET.iterparse(z.open(target),events=("end",)):
|
|
if not item.tag.endswith("}row"):
|
|
continue
|
|
rowno=int(item.attrib["r"])
|
|
cells={re.sub(r"\d","",c.attrib["r"]):value(c) for c in item}
|
|
if rowno==1:
|
|
headers=cells;item.clear();continue
|
|
raw={c:cells.get(c,"") for c in "ABCDEFGHIJKLMNO"}
|
|
if any(text(v) for v in raw.values()):
|
|
if any("\x00" in v for v in raw.values()):
|
|
raise ValueError("NUL in workbook; no lossy conversion allowed")
|
|
rows.append((rowno,raw))
|
|
item.clear()
|
|
if any(headers.get(c)!=h for c,h in {"H":"提问时间","I":"提问人","J":"原始问题"}.items()):
|
|
raise ValueError("approved event columns changed")
|
|
return headers,rows
|
|
|
|
def prepare(args,plan,decisions):
|
|
_,rows=read_rows(args.workbook,plan["source_sha256"])
|
|
by_row={r["excel_row"]:r for r in decisions}
|
|
if len(by_row)!=len(decisions) or len(rows)!=plan["data_rows"] or set(by_row)!={r for r,_ in rows}:
|
|
raise ValueError("approved decisions do not cover exactly the input rows")
|
|
observed=dt.datetime.now(TZ).isoformat()
|
|
for number,raw in rows:
|
|
decision=by_row[number]
|
|
if decision["status"] not in STATUSES:
|
|
raise ValueError("unapproved or ambiguous row disposition")
|
|
asked=None
|
|
if text(raw["H"]):
|
|
asked=dt.datetime.fromisoformat(text(raw["H"]).replace("/","-").replace("Z","+00:00"))
|
|
asked=asked.replace(tzinfo=TZ) if asked.tzinfo is None else asked.astimezone(TZ)
|
|
if asked.microsecond:
|
|
raise ValueError("timestamp precision differs from the approved seconds-only input")
|
|
key=[asked.strftime("%Y-%m-%d %H:%M:%S") if asked else "",text(raw["I"]),text(raw["J"])]
|
|
key_hash=sha(json.dumps(key,ensure_ascii=False).encode())
|
|
if key_hash!=decision["key_sha256"]:
|
|
raise ValueError("approved event key changed")
|
|
record_id="legacy_"+plan["source_sha256"]+"_"+str(number)
|
|
fields={k:raw[c] for k,c in {"question_chinese":"A","coverage":"C","is_correct":"D","product_category":"E","problem_type":"F","root_cause":"G","language":"K"}.items()}
|
|
payload={"id":record_id,"status":"historical_import","question":raw["J"],"answer":raw["B"],"asked_at":asked.isoformat() if asked else None,"time_source":"historical_excel","asker":raw["I"],"trigger_person":raw["L"],"department":raw["M"],"region":raw["N"],"references":[],"media":[],"source_coverage":{"verdict":"","reason_code":"","evaluator_version":""},"feedback":{"asker_value":"","up":0,"down":0},"master_fields":fields,"_identity":{},"_quality_observed_at":observed}
|
|
payload["_quality_hash"]=sha(json.dumps({"id":record_id,"question":raw["J"],"answer":raw["B"],"status":payload["status"],"references":[],"media":[]},ensure_ascii=False,sort_keys=True).encode())
|
|
yield {"row_number":number,"status":decision["status"],"event_key":key_hash,"expected_matches":decision["matched_record_sha256"],"asked_at":payload["asked_at"],"raw":raw,"payload":payload}
|
|
|
|
def build_sql(args,plan,stage,batch):
|
|
tenant=literal(args.tenant);digest=literal(plan["source_sha256"]);sheet=literal(plan["sheet"]);batchsql=literal(batch)
|
|
counts=(plan["data_rows"],plan["new_import_candidate_rows"],plan["existing_2_0_duplicate_rows"],plan["missing_key_review_rows"])
|
|
sql="BEGIN ISOLATION LEVEL SERIALIZABLE; SET LOCAL statement_timeout='120s'; SET LOCAL lock_timeout='15s';\n"
|
|
sql+="SELECT pg_advisory_xact_lock(hashtextextended("+tenant+"||':qa-history-import',0));\n"
|
|
sql+="LOCK TABLE qa_ops_master_records IN SHARE ROW EXCLUSIVE MODE;\n"
|
|
sql+="CREATE TEMP TABLE incoming_json(raw jsonb) ON COMMIT DROP;\nCOPY incoming_json(raw) FROM STDIN;\n"
|
|
for item in stage:
|
|
line=json.dumps(item,ensure_ascii=False,separators=(",",":"))
|
|
sql+=line.replace("\\","\\\\").replace("\n","\\n").replace("\r","\\r").replace("\t","\\t")+"\n"
|
|
sql+="\\.\nCREATE TEMP TABLE incoming ON COMMIT DROP AS SELECT (raw->>'row_number')::int row_number,raw->>'status' status,raw->>'event_key' event_key,raw->'expected_matches' expected_matches,(raw->>'asked_at')::timestamptz asked_at,raw->'raw' original_fields,raw->'payload' payload FROM incoming_json; CREATE INDEX ON incoming(event_key); ANALYZE incoming; ANALYZE qa_ops_master_records;\nCREATE TEMP TABLE admission_status(replayed boolean NOT NULL,inserted integer NOT NULL) ON COMMIT DROP;\n"
|
|
sql+="DO $import$ DECLARE replay boolean; bad bigint; BEGIN\n"
|
|
sql+="IF (SELECT max(version) FROM schema_migrations)<>53 THEN RAISE EXCEPTION 'unverified governance schema'; END IF;\n"
|
|
sql+="IF EXISTS(SELECT 1 FROM incoming WHERE status<>'missing_key_review' AND event_key<>qa_ops_master_event_key((asked_at)::timestamptz,original_fields->>'I',original_fields->>'J')) THEN RAISE EXCEPTION 'SQL and approved Python event identities differ'; END IF;\n"
|
|
sql+="SELECT EXISTS(SELECT 1 FROM qa_ops_master_import_batches WHERE tenant_key="+tenant+" AND source_sha256="+digest+" AND sheet_name="+sheet+") INTO replay;\nIF replay THEN\n"
|
|
sql+="IF NOT EXISTS(SELECT 1 FROM qa_ops_master_import_batches WHERE tenant_key="+tenant+" AND id="+batchsql+" AND ROW(total_rows,imported_rows,skipped_rows,held_rows)=ROW("+",".join(str(n)for n in counts)+")) THEN RAISE EXCEPTION 'existing import manifest differs'; END IF;\n"
|
|
sql+="IF (SELECT count(*) FROM qa_ops_master_import_rows WHERE tenant_key="+tenant+" AND batch_id="+batchsql+")<>"+str(counts[0])+" OR EXISTS(SELECT 1 FROM incoming i LEFT JOIN qa_ops_master_import_rows r ON r.tenant_key="+tenant+" AND r.batch_id="+batchsql+" AND r.row_number=(i.row_number)::int WHERE r.original_fields IS DISTINCT FROM i.original_fields OR r.disposition IS DISTINCT FROM CASE i.status WHEN 'new_import_candidate' THEN 'imported' WHEN 'existing_2_0_duplicate' THEN 'skipped_duplicate' ELSE 'held_missing_key' END) THEN RAISE EXCEPTION 'existing row audit differs'; END IF;\n"
|
|
sql+="INSERT INTO admission_status VALUES(true,0);\nELSE\n"
|
|
sql+="CREATE TEMP TABLE master_before ON COMMIT DROP AS SELECT id,row_to_json(m)::text AS contents FROM qa_ops_master_records m WHERE tenant_key="+tenant+";\n"
|
|
sql+="CREATE TEMP TABLE source_quality_before ON COMMIT DROP AS SELECT 'records' AS kind,to_jsonb(q) AS value FROM qa_ops_quality_records q WHERE tenant_key="+tenant+" UNION ALL SELECT 'inputs',to_jsonb(q) FROM qa_ops_quality_inputs q WHERE tenant_key="+tenant+" UNION ALL SELECT 'checks',to_jsonb(q) FROM qa_ops_quality_checks q WHERE tenant_key="+tenant+" UNION ALL SELECT 'reviews',to_jsonb(q) FROM qa_ops_quality_reviews q WHERE tenant_key="+tenant+";\n"
|
|
sql+="IF EXISTS(SELECT 1 FROM incoming i WHERE i.status='new_import_candidate' AND EXISTS(SELECT 1 FROM qa_ops_master_records m WHERE m.tenant_key="+tenant+" AND m.event_key=i.event_key)) OR EXISTS(SELECT 1 FROM incoming i WHERE i.status='existing_2_0_duplicate' AND (SELECT COALESCE(jsonb_agg(encode(sha256(convert_to(m.id,'UTF8')),'hex') ORDER BY m.id),'[]'::jsonb) FROM qa_ops_master_records m WHERE m.tenant_key="+tenant+" AND m.event_key=i.event_key) IS DISTINCT FROM i.expected_matches) THEN RAISE EXCEPTION 'comparison changed; refresh the preview before writing'; END IF;\n"
|
|
sql+="INSERT INTO qa_ops_master_import_batches(tenant_key,id,source_sha256,source_name,sheet_name,total_rows,imported_rows,skipped_rows,held_rows) VALUES("+tenant+","+batchsql+","+digest+","+literal(args.workbook.name)+","+sheet+","+",".join(str(n)for n in counts)+");\n"
|
|
sql+="INSERT INTO qa_ops_master_records(tenant_key,id,event_key,origin,asked_at,question,status,identity,quality_hash,payload,original_fields,batch_id,source_row) SELECT "+tenant+",payload->>'id',event_key,'historical_excel',(asked_at)::timestamptz,original_fields->>'J','historical_import','{}'::jsonb,payload->>'_quality_hash',payload,original_fields,"+batchsql+",(row_number)::int FROM incoming WHERE status='new_import_candidate';\n"
|
|
sql+="GET DIAGNOSTICS bad=ROW_COUNT; IF bad<>"+str(counts[1])+" THEN RAISE EXCEPTION 'imported row count differs'; END IF;\n"
|
|
sql+="INSERT INTO qa_ops_master_import_rows(tenant_key,batch_id,row_number,disposition,original_fields,master_id) SELECT "+tenant+","+batchsql+",(i.row_number)::int,CASE i.status WHEN 'new_import_candidate' THEN 'imported' WHEN 'existing_2_0_duplicate' THEN 'skipped_duplicate' ELSE 'held_missing_key' END,i.original_fields,CASE WHEN i.status='missing_key_review' THEN NULL ELSE m.id END FROM incoming i LEFT JOIN qa_ops_master_records m ON m.tenant_key="+tenant+" AND m.event_key=i.event_key AND i.status<>'missing_key_review';\n"
|
|
sql+="INSERT INTO qa_ops_master_source_links(tenant_key,source_kind,source_id,master_id) SELECT "+tenant+",'historical_excel',"+digest+"||':'||"+sheet+"||':'||row_number,master_id FROM qa_ops_master_import_rows WHERE tenant_key="+tenant+" AND batch_id="+batchsql+" AND master_id IS NOT NULL;\n"
|
|
sql+="IF EXISTS(SELECT 1 FROM master_before b LEFT JOIN qa_ops_master_records m ON m.tenant_key="+tenant+" AND m.id=b.id WHERE row_to_json(m)::text IS DISTINCT FROM b.contents) THEN RAISE EXCEPTION 'existing master record changed'; END IF;\n"
|
|
sql+="IF EXISTS((SELECT kind,value FROM source_quality_before EXCEPT SELECT kind,value FROM (SELECT 'records' kind,to_jsonb(q) value FROM qa_ops_quality_records q WHERE tenant_key="+tenant+" UNION ALL SELECT 'inputs',to_jsonb(q) FROM qa_ops_quality_inputs q WHERE tenant_key="+tenant+" UNION ALL SELECT 'checks',to_jsonb(q) FROM qa_ops_quality_checks q WHERE tenant_key="+tenant+" UNION ALL SELECT 'reviews',to_jsonb(q) FROM qa_ops_quality_reviews q WHERE tenant_key="+tenant+") current_source)) THEN RAISE EXCEPTION 'original source evaluation changed'; END IF;\n"
|
|
sql+="INSERT INTO admission_status VALUES(false,"+str(counts[1])+");\nEND IF;\nEND $import$;\n"
|
|
sql+="SELECT jsonb_build_object('state','IMPORT_TRANSACTION_VERIFIED','batch_id',"+batchsql+",'inserted_rows',inserted,'skipped_rows',"+str(counts[2])+",'held_rows',"+str(counts[3])+",'replayed',replayed,'master_total',(SELECT count(*) FROM qa_ops_master_records WHERE tenant_key="+tenant+"),'production_source_connected',false) FROM admission_status;\nCOMMIT;\n"
|
|
return sql
|
|
|
|
def main():
|
|
os.umask(0o077)
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
|
for name in ["workbook","decisions","plan","backup-verification","evidence-directory"]:
|
|
parser.add_argument("--"+name,type=Path,required=True)
|
|
parser.add_argument("--tenant",required=True)
|
|
parser.add_argument("--database",required=True)
|
|
parser.add_argument("--apply",action="store_true",required=True)
|
|
parser.add_argument("--live-apply",action="store_true")
|
|
args=parser.parse_args()
|
|
if not (re.fullmatch(r"kg_history_import_test_[a-f0-9]{12}",args.database) or (args.database=="governance" and args.live_apply)):
|
|
raise ValueError("only an isolated history test DB or explicitly approved local governance DB is permitted")
|
|
if not args.tenant or len(args.tenant)>128 or any(ord(c)<32 for c in args.tenant):
|
|
raise ValueError("invalid fixed governance tenant")
|
|
plan=json.loads(args.plan.read_text());decisions=json.loads(args.decisions.read_text())
|
|
expected=plan["source_sha256"]
|
|
if not re.fullmatch(r"[a-f0-9]{64}",expected) or sha(args.workbook.read_bytes())!=expected:
|
|
raise ValueError("approved input SHA differs")
|
|
if plan["sheet"]!="问题流水表" or plan["dedup_key"]!=["提问时间","提问人","原始问题"] or plan["within_excel_same_key_extra_rows"] or plan["ambiguous_existing_key_rows"]:
|
|
raise ValueError("unapproved matching contract")
|
|
approved=collections.Counter(d["status"] for d in decisions)
|
|
if approved!={"new_import_candidate":plan["new_import_candidate_rows"],"existing_2_0_duplicate":plan["existing_2_0_duplicate_rows"],"missing_key_review":plan["missing_key_review_rows"]}:
|
|
raise ValueError("approved row counts differ")
|
|
backup=json.loads(args.backup_verification.read_text());dump=args.backup_verification.parent/backup["backup_file"]
|
|
if backup["state"]!="BACKUP_RESTORE_VERIFIED" or not backup["real_restore_completed"] or not backup["all_table_row_counts_equal"] or not backup["all_qa_quality_org_frozen_content_hashes_equal"] or backup["source_workbook_sha256"]!=expected or dump.stat().st_size!=backup["backup_bytes"] or sha(dump.read_bytes())!=backup["backup_sha256"]:
|
|
raise ValueError("verified native backup gate failed")
|
|
args.evidence_directory.mkdir(mode=0o700,parents=True,exist_ok=True)
|
|
stage=list(prepare(args,plan,decisions));batch="xlsx_"+expected+"_"+sha(plan["sheet"].encode())[:12]
|
|
sql=build_sql(args,plan,stage,batch)
|
|
private=args.evidence_directory/("import-"+uuid.uuid4().hex[:12]+"-private.log")
|
|
result=subprocess.run(["docker","exec","-i",PG,"psql","-X","-qAt","-v","ON_ERROR_STOP=1","-U","governance","-d",args.database],input=sql,text=True,capture_output=True,timeout=150)
|
|
private.write_text(result.stderr)
|
|
if result.returncode:
|
|
raise RuntimeError("governance import has no verified commit receipt; read the batch before retrying; the transaction is all-or-nothing; inspect the restricted evidence log")
|
|
reports=[json.loads(line) for line in result.stdout.splitlines() if line.startswith('{')]
|
|
if len(reports)!=1:
|
|
raise RuntimeError("import receipt unavailable; inspect batch read-back before any retry")
|
|
report=reports[0]
|
|
# A committed receipt is checked by an independent connection, not just COMMIT's exit code.
|
|
query="BEGIN READ ONLY; SELECT jsonb_build_object('total',count(*),'imported',count(*)FILTER(WHERE disposition='imported'),'skipped',count(*)FILTER(WHERE disposition='skipped_duplicate'),'held',count(*)FILTER(WHERE disposition='held_missing_key')) FROM qa_ops_master_import_rows WHERE tenant_key="+literal(args.tenant)+" AND batch_id="+literal(batch)+"; COMMIT;"
|
|
check=subprocess.run(["docker","exec",PG,"psql","-X","-qAt","-v","ON_ERROR_STOP=1","-U","governance","-d",args.database,"-c",query],capture_output=True,text=True,timeout=30)
|
|
if check.returncode:
|
|
raise RuntimeError("import committed but independent audit read-back failed; do not claim complete")
|
|
actual=json.loads(check.stdout.strip())
|
|
if actual!={"total":plan["data_rows"],"imported":plan["new_import_candidate_rows"],"skipped":plan["existing_2_0_duplicate_rows"],"held":plan["missing_key_review_rows"]}:
|
|
raise RuntimeError("committed audit row counts differ")
|
|
if sha(args.workbook.read_bytes())!=expected:
|
|
raise RuntimeError("input changed during import")
|
|
report.update(independent_read_back_verified=True,source_unchanged=True)
|
|
evidence=args.evidence_directory/("import-"+uuid.uuid4().hex[:12]+"-verification.json")
|
|
evidence.write_text(json.dumps(report,ensure_ascii=False,indent=2))
|
|
print(json.dumps(report,ensure_ascii=False))
|
|
|
|
if __name__=="__main__":
|
|
main()
|