You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
326 lines
13 KiB
326 lines
13 KiB
# -*- coding: utf-8 -*-
|
|
r"""票 06：A4 客户管理 Bruno 文档生成器（gen + inject）。
|
|
- 结构听源码：目录=源码 @Tag 原文；docs 首行=对账钥匙 METHOD 路径
|
|
- 示例值注入：specimens-core.json / specimens-heavy.json 实测真值（响应示例全真值，
|
|
请求示例值优先实测，缺失按语义兜底）
|
|
- 输出 D:\code\crm-api-docs（python 直写，Write 工具不可出工作区）
|
|
- 幂等：重跑整目录重建（A4 为全新生成，无「人的手改」需保护）
|
|
"""
|
|
import io, sys, os, re, json
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
|
sys.path.insert(0, r'd:\code\crm-backend-matt\.scratch\customer-e2e')
|
|
from defs_a4 import DEFS, PATH_VARS, SEMANTIC
|
|
|
|
DOCS = r'D:\code\crm-api-docs'
|
|
BASE = os.path.join(DOCS, 'A4 客户管理')
|
|
SP_DIR = r'd:\code\crm-backend-matt\.scratch\customer-e2e'
|
|
|
|
spec = {}
|
|
for f in ('specimens-core.json', 'specimens-heavy.json'):
|
|
spec.update(json.load(open(os.path.join(SP_DIR, f), encoding='utf-8')))
|
|
|
|
|
|
def norm_key(path):
|
|
"""specimens 钥匙归一：15+ 位数字段与 taskId 段 → {id}；workspace 字面值展开。"""
|
|
p = re.sub(r'/\d{15,}', '/{id}', path)
|
|
p = re.sub(r'/\d{6,14}', '/{id}', p) if '/import/' in path else p
|
|
return p
|
|
|
|
|
|
def spec_lookup(method, path_tpl):
|
|
"""候选钥匙：workspace 三字面值 + 全占位归一 {id}。"""
|
|
cands = []
|
|
ws = '{workspace}' in path_tpl
|
|
bases = [path_tpl.replace('{workspace}', w) for w in ('mine', 'overview', 'pool')] if ws else [path_tpl]
|
|
for b in bases:
|
|
cands.append((method, b))
|
|
cands.append((method, re.sub(r'\{(id|customerId|memberUserId|taskId)\}', '{id}', b)))
|
|
for m, p in cands:
|
|
if f'{m} {p}' in spec:
|
|
return spec[f'{m} {p}']
|
|
return None
|
|
|
|
|
|
def url_of(path_tpl):
|
|
for k, v in PATH_VARS.items():
|
|
path_tpl = path_tpl.replace(k, v)
|
|
return path_tpl.replace('{taskId}', str(_task_id()))
|
|
|
|
|
|
_TASKID = None
|
|
def _task_id():
|
|
global _TASKID
|
|
if _TASKID is None:
|
|
d = (spec.get('POST /api/customer/import/upload') or {}).get('response', {}).get('data') or {}
|
|
_TASKID = d.get('taskId') or '750854700000000001'
|
|
return _TASKID
|
|
|
|
|
|
def jtype(v):
|
|
if v is None: return 'string'
|
|
if isinstance(v, bool): return 'boolean'
|
|
if isinstance(v, (int, float)): return 'number'
|
|
if isinstance(v, list): return 'array'
|
|
return 'string'
|
|
|
|
|
|
def sample(v, maxlen=48):
|
|
s = json.dumps(v, ensure_ascii=False) if not isinstance(v, str) else v
|
|
s = s if len(s) <= maxlen else s[:maxlen] + '…'
|
|
return s
|
|
|
|
|
|
def resp_struct_table(data, indent=''):
|
|
"""从实测 data 推导字段表（字段/类型/示例值）；分页 content 展开一层。"""
|
|
rows = []
|
|
if isinstance(data, dict):
|
|
for k, v in data.items():
|
|
if k == 'content' and isinstance(v, list) and v and isinstance(v[0], dict):
|
|
rows.append(f'| content[] | array | 分页元素，实测共 {len(v)} 条，结构如下 |')
|
|
for k2, v2 in v[0].items():
|
|
rows.append(f'| content[].{k2} | {jtype(v2)} | {sample(v2)} |')
|
|
else:
|
|
rows.append(f'| {k} | {jtype(v)} | {sample(v)} |')
|
|
elif isinstance(data, list) and data and isinstance(data[0], dict):
|
|
for k, v in data[0].items():
|
|
rows.append(f'| {indent}[].{k} | {jtype(v)} | {sample(v)} |')
|
|
elif isinstance(data, list):
|
|
rows.append(f'| {indent}[] | {jtype(data[0] if data else None)} | {sample(data[0] if data else "", 24)} |')
|
|
return rows
|
|
|
|
|
|
def trim_spec_resp(d):
|
|
"""响应示例裁剪：content 取 1 条，其余原样。"""
|
|
if not isinstance(d, dict):
|
|
return d, ''
|
|
note = ''
|
|
dd = json.loads(json.dumps(d)) # deep copy
|
|
data = dd.get('data')
|
|
if isinstance(data, dict) and isinstance(data.get('content'), list) and len(data['content']) > 1:
|
|
note = f"（实测 total={data.get('total', '?')}，content 截取 1 条）"
|
|
data['content'] = data['content'][:1]
|
|
elif isinstance(data, list) and len(data) > 3:
|
|
note = f'（实测 {len(data)} 条，截取前 3 条）'
|
|
dd['data'] = data[:3]
|
|
return dd, note
|
|
|
|
|
|
def gen_bru(d, seq):
|
|
method = d['method'].lower()
|
|
key = f"{d['method']} {re.sub(chr(92)+'{(id|customerId|memberUserId|taskId)}', '{id}', d['path'])}"
|
|
sp = spec_lookup(d['method'], d['path'])
|
|
url = url_of(d['path'])
|
|
L = []
|
|
L.append('meta {')
|
|
L.append(f' name: {d["name"]}')
|
|
L.append(' type: http')
|
|
L.append(f' seq: {seq}')
|
|
L.append(' tags: [')
|
|
L.append(' generated')
|
|
L.append(' ]')
|
|
L.append('}')
|
|
L.append('')
|
|
body_kind = {'form': 'formUrlEncoded', 'none': 'none', 'json': 'json', 'multipart': 'multipartFormData'}[d['body']]
|
|
L.append(f'{method} {{')
|
|
L.append(f' url: {{{{baseUrl}}}}{url}')
|
|
L.append(f' body: {body_kind}')
|
|
L.append(' auth: inherit')
|
|
L.append('}')
|
|
L.append('')
|
|
|
|
# ---- 参数块 ----
|
|
qp = d['params'] if d['body'] == 'none' else []
|
|
fp = d['params'] if d['body'] in ('form', 'multipart') else []
|
|
if qp:
|
|
L.append('params:query {')
|
|
L.append('')
|
|
for n, t, r, _ in qp:
|
|
L.append(f' {n}: {SEMANTIC.get(n, "") if SEMANTIC.get(n, "") != "" else "1"}')
|
|
L.append('}')
|
|
L.append('')
|
|
if d['body'] == 'form' and fp:
|
|
L.append('body:form-urlencoded {')
|
|
L.append('')
|
|
spb = ((sp or {}).get('request') or {}).get('body')
|
|
for n, t, r, _ in fp:
|
|
v = (spb or {}).get(n) if isinstance(spb, dict) else None
|
|
if v is None or v == '':
|
|
sv = SEMANTIC.get(n, '')
|
|
if sv == '' and r:
|
|
sv = '1'
|
|
v = sv
|
|
L.append(f' {n}: {v}')
|
|
L.append('}')
|
|
L.append('')
|
|
if d['body'] == 'multipart':
|
|
L.append('body:multipartFormData {')
|
|
L.append('')
|
|
L.append(' file: @客户导入模板.xlsx')
|
|
L.append(f" importMode: {SEMANTIC['importMode']}")
|
|
L.append('}')
|
|
L.append('')
|
|
if d['body'] == 'json':
|
|
L.append('body:json {')
|
|
L.append('')
|
|
spb = ((sp or {}).get('request') or {}).get('body')
|
|
body_json = spb if isinstance(spb, dict) and spb else {n: int(SEMANTIC.get(n, 0) or 0) for n, t, r, _ in fp}
|
|
for ln in json.dumps(body_json, ensure_ascii=False, indent=2).split('\n'):
|
|
L.append(' ' + ln)
|
|
L.append('}')
|
|
L.append('')
|
|
|
|
# ---- docs ----
|
|
L.append('docs {')
|
|
L.append(f' # {d["name"]}')
|
|
L.append(f' `{d["method"]} {d["path"]}`（本行兼作对账钥匙，勿手改）')
|
|
for para in d['desc'].split('；') if False else [d['desc']]:
|
|
L.append(f' {para}')
|
|
L.append('')
|
|
L.append(' ## 请求参数')
|
|
L.append('')
|
|
if d['params']:
|
|
L.append(' | 参数 | 类型 | 必填 | 说明 |')
|
|
L.append(' | --- | --- | --- | --- |')
|
|
for n, t, r, desc in d['params']:
|
|
L.append(f' | {n} | {t} | {"是" if r else "否"} | {desc} |')
|
|
else:
|
|
L.append(' 无（路径变量外无业务参数）')
|
|
L.append('')
|
|
L.append(' ## 请求示例')
|
|
L.append('')
|
|
if d['body'] == 'none':
|
|
q = '&'.join(f'{n}={SEMANTIC.get(n, "1")}' for n, t, r, _ in d['params'] if SEMANTIC.get(n, '1') != '')
|
|
shown = d['path']
|
|
for k, v in PATH_VARS.items():
|
|
shown = shown.replace(k, v)
|
|
shown = shown.replace('{taskId}', str(_task_id()))
|
|
L.append(' 请求头：Authorization: Bearer {{token}}')
|
|
L.append('')
|
|
L.append(f' {d["method"]} {shown}' + (f'?{q}' if q else ''))
|
|
elif d['body'] == 'form':
|
|
L.append(' 请求头：Authorization: Bearer {{token}}')
|
|
L.append(' Content-Type: application/x-www-form-urlencoded')
|
|
L.append('')
|
|
shown = d['path']
|
|
for k, v in PATH_VARS.items():
|
|
shown = shown.replace(k, v)
|
|
L.append(f' {d["method"]} {shown}')
|
|
pairs = []
|
|
spb = ((sp or {}).get('request') or {}).get('body')
|
|
for n, t, r, _ in d['params']:
|
|
v = (spb or {}).get(n) if isinstance(spb, dict) else None
|
|
if v is None or v == '':
|
|
v = SEMANTIC.get(n, '')
|
|
if v == '':
|
|
continue
|
|
pairs.append(f'{n}={v}')
|
|
L.append(' // 示例数据（E2E 实测值注入），非真实返回' if sp else ' // 示例数据，非真实返回')
|
|
L.append(' ' + '&'.join(pairs))
|
|
elif d['body'] == 'multipart':
|
|
L.append(' 请求头：Authorization: Bearer {{token}}')
|
|
L.append(' Content-Type: multipart/form-data')
|
|
L.append('')
|
|
L.append(f' {d["method"]} {d["path"]}')
|
|
L.append(' file=@客户导入模板.xlsx')
|
|
L.append(f" importMode={SEMANTIC['importMode']}")
|
|
elif d['body'] == 'json':
|
|
L.append(' 请求头：Authorization: Bearer {{token}}')
|
|
L.append(' Content-Type: application/json')
|
|
L.append('')
|
|
L.append(f' {d["method"]} {d["path"]}')
|
|
L.append(' ```json')
|
|
spb = ((sp or {}).get('request') or {}).get('body')
|
|
body_json = spb if isinstance(spb, dict) and spb else {n: int(SEMANTIC.get(n, 0) or 0) for n, t, r, _ in d['params']}
|
|
for ln in json.dumps(body_json, ensure_ascii=False, indent=2).split('\n'):
|
|
L.append(' ' + ln)
|
|
L.append(' ```')
|
|
L.append('')
|
|
L.append(' ## 响应 data 结构')
|
|
L.append('')
|
|
rdata = ((sp or {}).get('response') or {}).get('data') if sp else None
|
|
if rdata is not None:
|
|
rows = resp_struct_table(rdata)
|
|
if rows:
|
|
L.append(' | 字段 | 类型 | 示例值（E2E 实测） |')
|
|
L.append(' | --- | --- | --- |')
|
|
for r in rows:
|
|
L.append(' ' + r)
|
|
else:
|
|
L.append(f' {d["resp"]}')
|
|
else:
|
|
L.append(f' {d["resp"]}')
|
|
if d['errs']:
|
|
L.append('')
|
|
L.append(' ## 错误码')
|
|
L.append('')
|
|
for e in d['errs']:
|
|
L.append(f' - {e}')
|
|
L.append('')
|
|
L.append(' ## 响应示例')
|
|
L.append('')
|
|
if sp and (sp.get('response') or {}).get('data', 'MISS') != 'MISS':
|
|
trimmed, note = trim_spec_resp(sp['response'])
|
|
L.append(' // 示例数据（E2E 实测真值），字段结构以上表为准' + note)
|
|
L.append(' ```json')
|
|
for ln in json.dumps(trimmed, ensure_ascii=False, indent=2).split('\n'):
|
|
L.append(' ' + ln)
|
|
L.append(' ```')
|
|
elif d['body'] == 'none' and rdata is None and not d['resp'].startswith('非信封') and '文字' not in d['resp']:
|
|
L.append(' // 示例数据，字段结构以上表为准，非真实返回')
|
|
L.append(' ```json')
|
|
for ln in json.dumps({'code': 0, 'success': True, 'message': 'success', 'data': None}, indent=2).split('\n'):
|
|
L.append(' ' + ln)
|
|
L.append(' ```')
|
|
else:
|
|
L.append(f' {d["resp"]}')
|
|
L.append('}')
|
|
return '\n'.join(L) + '\n'
|
|
|
|
|
|
def folder_bru(name, seq):
|
|
return f'meta {{\n name: {name}\n seq: {seq}\n}}\n'
|
|
|
|
|
|
def main():
|
|
# 按 dir 分组；dir='A4 客户管理/xxx' 相对 BASE（剥首段），其余相对 collection 根
|
|
groups = {}
|
|
roots = {}
|
|
for d in DEFS:
|
|
g = d['dir']
|
|
if g.startswith('A4 客户管理/'):
|
|
roots[g] = BASE
|
|
rel = g[len('A4 客户管理/'):]
|
|
else:
|
|
roots[g] = DOCS
|
|
rel = g
|
|
groups.setdefault((roots[g], rel), []).append(d)
|
|
written = []
|
|
for (root, rel), items in groups.items():
|
|
gdir = os.path.join(root, *rel.split('/'))
|
|
os.makedirs(gdir, exist_ok=True)
|
|
short = rel.split('/')[-1]
|
|
open(os.path.join(gdir, 'folder.bru'), 'w', encoding='utf-8', newline='').write(folder_bru(short, 1))
|
|
# 中间层级 folder.bru（如 根级 客户管理/）
|
|
parent = gdir
|
|
while os.path.relpath(parent, root) not in ('.', ''):
|
|
parent = os.path.dirname(parent)
|
|
for missing in [gdir]:
|
|
pass
|
|
if root == DOCS: # 根级目录需要逐级 folder.bru
|
|
acc = DOCS
|
|
for seg in rel.split('/')[:-1]:
|
|
acc = os.path.join(acc, seg)
|
|
fp = os.path.join(acc, 'folder.bru')
|
|
if not os.path.exists(fp):
|
|
open(fp, 'w', encoding='utf-8', newline='').write(folder_bru(seg, 12))
|
|
for i, d in enumerate(items, 1):
|
|
p = os.path.join(gdir, d['name'] + '.bru')
|
|
open(p, 'w', encoding='utf-8', newline='').write(gen_bru(d, i))
|
|
written.append(os.path.relpath(p, DOCS))
|
|
print(f'generated {len(written)} .bru in {len(groups)} dirs')
|
|
for w in written:
|
|
print(' ', w)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|