193 lines
7.5 KiB
Python
193 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Write all P4 rewrite records to bitable using direct API calls.
|
|
Avoids shell escaping issues.
|
|
"""
|
|
import json, subprocess, os, sys
|
|
|
|
APP_TOKEN = "CMHSbUUjka3TrUsaxxEc297ongf"
|
|
TABLE_ID = "tblVmeDtBDKsAEfz"
|
|
BASE_DIR = os.path.expanduser("~/.openclaw/workspace-xiaoyan")
|
|
CRED_FILE = os.path.expanduser("~/.openclaw/credentials/xiaoyan/config.json")
|
|
|
|
RECORD_IDS = {
|
|
"021301": "recvjufKJO8d0O",
|
|
"021401": "recvjufM763ijb",
|
|
"021501": "recvjufM76lEsW",
|
|
"021601": "recvjufM76hNv5",
|
|
"021701": "recvjufM76eMKs",
|
|
"021801": "recvjufM76lRHQ",
|
|
"022101": "recvjufM76o6of",
|
|
"032501": "recvjufM76k4dx",
|
|
"032901": "recvjufM76frUP",
|
|
}
|
|
|
|
def get_token():
|
|
with open(CRED_FILE) as f:
|
|
cred = json.load(f)
|
|
app_id = cred["apps"][0]["appId"]
|
|
app_secret = cred["apps"][0]["appSecret"]
|
|
result = subprocess.run([
|
|
"curl", "-s", "-X", "POST",
|
|
"https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
|
|
"-H", "Content-Type: application/json",
|
|
"-d", json.dumps({"app_id": app_id, "app_secret": app_secret})
|
|
], capture_output=True, text=True)
|
|
return json.loads(result.stdout)["tenant_access_token"]
|
|
|
|
def update_record(token, record_id, fields_dict):
|
|
"""Update a record via curl"""
|
|
# Write fields to temp file to avoid command-line escaping
|
|
tmp_file = f"/tmp/p4_update_{record_id}.json"
|
|
payload = json.dumps({"fields": fields_dict}, ensure_ascii=False)
|
|
with open(tmp_file, 'w') as f:
|
|
f.write(payload)
|
|
|
|
url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records/{record_id}"
|
|
result = subprocess.run([
|
|
"curl", "-s", "-X", "PUT", url,
|
|
"-H", f"Authorization: Bearer {token}",
|
|
"-H", "Content-Type: application/json",
|
|
"-d", f"@{tmp_file}"
|
|
], capture_output=True, text=True, timeout=30)
|
|
|
|
os.remove(tmp_file)
|
|
return json.loads(result.stdout)
|
|
|
|
def verify_record(token, record_id):
|
|
"""Re-read a record to verify"""
|
|
url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records/{record_id}"
|
|
result = subprocess.run([
|
|
"curl", "-s", "-X", "GET", url,
|
|
"-H", f"Authorization: Bearer {token}"
|
|
], capture_output=True, text=True, timeout=30)
|
|
return json.loads(result.stdout)
|
|
|
|
def main():
|
|
mode = sys.argv[1] if len(sys.argv) > 1 else "test"
|
|
|
|
# Get token
|
|
print("Getting token...")
|
|
token = get_token()
|
|
print(f"Token: {token[:20]}...")
|
|
|
|
# Import the rewrite data
|
|
sys.path.insert(0, os.path.join(BASE_DIR, "scripts"))
|
|
import rewrite_p4_all as rw
|
|
|
|
from rewrite_p4_all import make_q1_text
|
|
|
|
results = []
|
|
|
|
for qsid, record_id in RECORD_IDS.items():
|
|
data = rw.RECORD_DATA.get(qsid)
|
|
if not data:
|
|
continue
|
|
|
|
first_block = data["first_block"]
|
|
second_block = data["second_block"]
|
|
first_qs = data["first_qs_list"]
|
|
second_qs = data["second_qs_list"]
|
|
|
|
# Clean dialogues from jsonData
|
|
for q in first_block["questionSet"]:
|
|
q.pop("dialogue", None)
|
|
if second_block and isinstance(second_block, dict) and "questionSet" in second_block:
|
|
for q in second_block["questionSet"]:
|
|
q.pop("dialogue", None)
|
|
|
|
json_data = {"first": first_block, "second": second_block}
|
|
|
|
q1_text = make_q1_text(first_qs)
|
|
q2_text = make_q1_text(second_qs) if second_qs else ""
|
|
|
|
fields = {
|
|
"jsonData": json.dumps(json_data, ensure_ascii=False),
|
|
"题目1": q1_text,
|
|
"题目2": q2_text,
|
|
}
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"QSID: {qsid} | Record: {record_id}")
|
|
print(f"First: {len(first_qs)} Qs, Second: {len(second_qs)} Qs")
|
|
|
|
if mode == "test":
|
|
# Test one record only
|
|
if qsid == "021301":
|
|
result = update_record(token, record_id, fields)
|
|
code = result.get("code", -1)
|
|
msg = result.get("msg", "")
|
|
if code == 0:
|
|
print(f" ✅ Test write SUCCESS")
|
|
# Verify
|
|
v = verify_record(token, record_id)
|
|
vcode = v.get("code", -1)
|
|
if vcode == 0:
|
|
rec = v.get("data", {}).get("record", {})
|
|
jd = rec.get("fields", {}).get("jsonData", "")
|
|
if jd:
|
|
try:
|
|
parsed = json.loads(jd)
|
|
fq = parsed.get("first", {}).get("questionSet", [])
|
|
sq = parsed.get("second", {}).get("questionSet", [])
|
|
print(f" ✅ Verify: {len(fq)} first Qs, {len(sq)} second Qs")
|
|
print(f" ✅ jsonData valid JSON")
|
|
except Exception as e:
|
|
print(f" ❌ jsonData parse error: {e}")
|
|
else:
|
|
print(f" ❌ jsonData field empty")
|
|
else:
|
|
print(f" ❌ Verify FAILED: {v.get('msg', '')}")
|
|
else:
|
|
print(f" ❌ Test write FAILED: {msg}")
|
|
break
|
|
|
|
elif mode == "write":
|
|
result = update_record(token, record_id, fields)
|
|
code = result.get("code", -1)
|
|
msg = result.get("msg", "")
|
|
if code == 0:
|
|
print(f" ✅ Write SUCCESS")
|
|
results.append((qsid, True, ""))
|
|
else:
|
|
print(f" ❌ Write FAILED: {msg}")
|
|
results.append((qsid, False, msg))
|
|
|
|
if mode == "write":
|
|
success = sum(1 for r in results if r[1])
|
|
fails = sum(1 for r in results if not r[1])
|
|
print(f"\n{'='*60}")
|
|
print(f"WRITE SUMMARY: {success} succeeded, {fails} failed")
|
|
for qsid, ok, msg in results:
|
|
status = "✅" if ok else "❌"
|
|
print(f" {status} {qsid}" + (f" - {msg}" if msg else ""))
|
|
|
|
# Verify all
|
|
print(f"\n{'='*60}")
|
|
print("VERIFYING ALL RECORDS...")
|
|
for qsid, record_id in RECORD_IDS.items():
|
|
v = verify_record(token, record_id)
|
|
vcode = v.get("code", -1)
|
|
if vcode == 0:
|
|
rec = v.get("data", {}).get("record", {})
|
|
jd = rec.get("fields", {}).get("jsonData", "")
|
|
q1 = rec.get("fields", {}).get("题目1", "")
|
|
q2 = rec.get("fields", {}).get("题目2", "")
|
|
if jd:
|
|
try:
|
|
parsed = json.loads(jd)
|
|
fq = parsed.get("first", {}).get("questionSet", [])
|
|
sq = parsed.get("second", {}).get("questionSet", [])
|
|
fqsid = parsed.get("first", {}).get("questionSetID", "")
|
|
sqsid = parsed.get("second", {}).get("questionSetID", "") if isinstance(parsed.get("second", {}), dict) else ""
|
|
print(f" ✅ {qsid}: {len(fq)} first Qs, {len(sq)} second Qs, QSID={fqsid}, 题目1={len(q1)}chars, 题目2={len(q2)}chars")
|
|
except Exception as e:
|
|
print(f" ❌ {qsid}: jsonData parse error: {e}")
|
|
else:
|
|
print(f" ❌ {qsid}: jsonData empty")
|
|
else:
|
|
print(f" ❌ {qsid}: verify failed: {v.get('msg', '')}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|