#!/usr/bin/env python3 """ customer_service.py - customer service skill helper Aligned with current server implementation: - `/service/pn_data_analysis`: `listProject`, `statisticCustomerServiceReplyForSkill` - `/service/skill_for_customer_service`: `getWaitingDraftTroubles`, `getWaitingDraftTroubleById`, `updateWorkOrderDraft`, `updateDraft`, `markWorkOrderError`, `markWorkOrderOperationHandled`, `updateReplyDraft`, `replyByDraftId`, `getTroubleDetailBySubjectId`, `getTroubleListByProject`, `getTroubleListForAllProjects`, `updateCustomerServiceTags`, `getOperationProcessingWorkOrderCount` """ import argparse import base64 import json import os import subprocess import sys import time import urllib.error import urllib.request from datetime import datetime from pathlib import Path from typing import Union API_PATH = "/service/pn_data_analysis" CUSTOMER_SKILL_PATH = "/service/skill_for_customer_service" ENV_TOKEN = "PN_DATA_ANALYSIS_TOKEN" DEFAULT_API_BASE = "https://pnv2.17995api.net" def _strip_quotes(value: str) -> str: value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: return value[1:-1] return value def _discover_project_root() -> Path: here = Path(__file__).resolve() for parent in [here.parent] + list(here.parents): if (parent / ".git").exists(): return parent return here.parent def _load_env_file(env_file: Path) -> None: if not env_file.exists() or not env_file.is_file(): return try: for raw in env_file.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[len("export ") :].strip() if "=" not in line: continue key, value = line.split("=", 1) key = key.strip() if not key: continue os.environ[key] = _strip_quotes(value) except OSError: return def _load_local_env() -> None: skill_dir = Path(__file__).resolve().parent root = _discover_project_root() candidates = [ skill_dir / ".env.customer-service", skill_dir / ".env", root / ".env.customer-service", root / ".env", ] for env_file in candidates: _load_env_file(env_file) _load_local_env() def _api_base() -> str: return os.environ.get("PN_DATA_ANALYSIS_BASE", DEFAULT_API_BASE).rstrip("/") def _get_token() -> str: token = (os.environ.get(ENV_TOKEN) or "").strip() if token == "": raise SystemExit( f"未配置 {ENV_TOKEN}。请在中台右上角生成数据分析密钥后写入环境变量。" ) return token def _log_http_error(e: urllib.error.HTTPError) -> None: print(f"HTTP {e.code} {e.reason}", file=sys.stderr) try: body = e.read().decode("utf-8", errors="replace")[:2000] except Exception: body = "" if body.strip(): print(body, file=sys.stderr) def _post_json_via_curl(path: str, payload: dict, timeout: int = 60) -> Union[dict, list]: url = f"{_api_base()}{path}" cmd = [ "curl", "-sS", "--fail-with-body", "-X", "POST", url, "-H", "Content-Type: application/json;charset=UTF-8", "-H", f"X-Pn-Token: {_get_token()}", "--max-time", str(timeout), "--data", json.dumps(payload, ensure_ascii=False), ] proc = subprocess.run(cmd, capture_output=True, text=True) if proc.returncode != 0: detail = (proc.stderr or proc.stdout or "").strip() raise RuntimeError(f"curl fallback failed ({proc.returncode}): {detail[:2000]}") return json.loads(proc.stdout) def _post_json(path: str, payload: dict, timeout: int = 60) -> Union[dict, list]: req = urllib.request.Request( f"{_api_base()}{path}", data=json.dumps(payload).encode("utf-8"), headers={ "Content-Type": "application/json;charset=UTF-8", "X-Pn-Token": _get_token(), }, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: if e.code >= 500: return _post_json_via_curl(path, payload, timeout=timeout) _log_http_error(e) raise except urllib.error.URLError: return _post_json_via_curl(path, payload, timeout=timeout) def _parse_json_object(raw: str, arg_name: str) -> dict: text = (raw or "").strip() if text == "": return {} try: data = json.loads(text) except json.JSONDecodeError as e: raise SystemExit(f"{arg_name} 不是合法 JSON: {e.msg}") from e if not isinstance(data, dict): raise SystemExit(f"{arg_name} 必须是 JSON 对象，例如 '{{\"status\": -1}}'") return data def decode_pn_data_analysis_token_payload(token: str) -> dict: token = (token or "").strip() parts = token.split(".", 1) if len(parts) != 2: raise ValueError("token 须为两段：base64(JSON).signature") payload_b64, _sig = parts pad = (-len(payload_b64)) % 4 raw = base64.b64decode(payload_b64 + ("=" * pad), validate=False) data = json.loads(raw.decode("utf-8")) if not isinstance(data, dict): raise ValueError("载荷不是 JSON 对象") return data def format_pn_token_payload(data: dict) -> str: uid = data.get("uid") exp = data.get("exp") lines = [f"uid（中台用户 id）: {uid}"] if exp is not None: try: exp_i = int(exp) lines.append(f"exp（过期 Unix 秒）: {exp_i}") lines.append( f"过期时间（本地）: {datetime.fromtimestamp(exp_i).strftime('%Y-%m-%d %H:%M:%S')}" ) lines.append("状态: 已过期" if exp_i < int(time.time()) else "状态: 未过期") except (TypeError, ValueError, OSError): lines.append(f"exp: {exp!r}") return "\n".join(lines) def query_project_list() -> Union[dict, list]: return _post_json(API_PATH, {"dataType": "listProject"}) def query_waiting_draft_troubles() -> Union[dict, list]: return _post_json(CUSTOMER_SKILL_PATH, {"api": "getWaitingDraftTroubles"}) def query_waiting_draft_trouble_by_id(subject_id: int) -> Union[dict, list]: return _post_json( CUSTOMER_SKILL_PATH, {"api": "getWaitingDraftTroubleById", "subjectId": int(subject_id)}, ) def query_update_work_order_draft( subject_id: int, title: str, content: str ) -> Union[dict, list]: return _post_json( CUSTOMER_SKILL_PATH, { "api": "updateWorkOrderDraft", "subjectId": int(subject_id), "title": title, "content": content, }, ) def query_update_draft( subject_id: int, work_order_title: str, work_order_content: str, reply_content: str, ai_source: str = "", ) -> Union[dict, list]: payload = { "api": "updateDraft", "subjectId": int(subject_id), "workOrderTitle": work_order_title, "workOrderContent": work_order_content, "replyContent": reply_content, } ai_source = (ai_source or "").strip() if ai_source != "": payload["aiSource"] = ai_source return _post_json(CUSTOMER_SKILL_PATH, payload) def query_mark_work_order_error(subject_id: int) -> Union[dict, list]: payload = { "api": "markWorkOrderError", "subjectId": int(subject_id), } return _post_json( CUSTOMER_SKILL_PATH, payload, ) def query_mark_work_order_operation_handled( subject_id: int, title: str, content: str ) -> Union[dict, list]: return _post_json( CUSTOMER_SKILL_PATH, { "api": "markWorkOrderOperationHandled", "subjectId": int(subject_id), "title": title, "content": content, }, ) def query_update_reply_draft( subject_id: int, content: str, ai_source: str = "" ) -> Union[dict, list]: payload = { "api": "updateReplyDraft", "subjectId": int(subject_id), "content": content, } ai_source = (ai_source or "").strip() if ai_source != "": payload["aiSource"] = ai_source return _post_json(CUSTOMER_SKILL_PATH, payload) def query_reply_by_draft_id( subject_id: int, draft_id: str, resolved: bool ) -> Union[dict, list]: return _post_json( CUSTOMER_SKILL_PATH, { "api": "replyByDraftId", "subjectId": int(subject_id), "draftId": draft_id, "resolved": "yes" if resolved else "no", }, timeout=120, ) def query_trouble_detail_by_subject_id(subject_id: int) -> Union[dict, list]: return _post_json( CUSTOMER_SKILL_PATH, { "api": "getTroubleDetailBySubjectId", "subjectId": int(subject_id), }, ) def query_trouble_list_by_project(pid: int, data: dict) -> Union[dict, list]: return _post_json( CUSTOMER_SKILL_PATH, { "api": "getTroubleListByProject", "pid": int(pid), "data": data, }, ) def query_trouble_list_for_all_projects(eid: int, data: dict) -> Union[dict, list]: return _post_json( CUSTOMER_SKILL_PATH, { "api": "getTroubleListForAllProjects", "eid": int(eid), "data": data, }, ) def query_operation_processing_work_order_count() -> Union[dict, list]: return _post_json( CUSTOMER_SKILL_PATH, { "api": "getOperationProcessingWorkOrderCount", }, ) def query_update_customer_service_tags( subject_id: int, tags: list[str] ) -> Union[dict, list]: if len(tags) > 3: raise SystemExit("--update-customer-service-tags 最多支持 3 个标签") return _post_json( CUSTOMER_SKILL_PATH, { "api": "updateCustomerServiceTags", "subjectId": int(subject_id), "tags": tags, }, ) def query_statistic_customer_service_reply_for_skill( time_span_start: int, time_span_end: int ) -> Union[dict, list]: return _post_json( API_PATH, { "dataType": "statisticCustomerServiceReplyForSkill", "timeSpan": [int(time_span_start), int(time_span_end)], }, timeout=120, ) def format_project_list_result(result: Union[dict, list]) -> str: if isinstance(result, list): return f"项目列表: 共 {len(result)} 个\n" + json.dumps( result, ensure_ascii=False, indent=2 ) return json.dumps(result, ensure_ascii=False, indent=2) def format_statistic_customer_service_reply_for_skill_result( result: Union[dict, list] ) -> str: data: Union[dict, list] = result if isinstance(result, dict): if "data" in result and isinstance(result["data"], (dict, list)): data = result["data"] list_rows: list = [] score_rows: list = [] if isinstance(data, list): list_rows = data elif isinstance(data, dict): if isinstance(data.get("list"), list): list_rows = data["list"] if isinstance(data.get("score"), list): score_rows = data["score"] else: return json.dumps(result, ensure_ascii=False, indent=2) if len(list_rows) > 0 and len(score_rows) > 0: for score_item in score_rows: if not isinstance(score_item, dict): continue gmid = score_item.get("gmid") for list_item in list_rows: if isinstance(list_item, dict) and list_item.get("gmid") == gmid: list_item["good"] = score_item.get("good") list_item["bad"] = score_item.get("bad") break total_reply = 0 total_ai_reply = 0 for row in list_rows: if not isinstance(row, dict): continue total_reply += int(row.get("reply") or 0) total_ai_reply += int(row.get("aireply") or 0) lines = [ "客服回复统计（口径：reply=人工回复数，aireply=采用AI回复数）:", f"- 人工回复总数(reply): {total_reply}", f"- AI回复采用总数(aireply): {total_ai_reply}", "JSON:", json.dumps(data, ensure_ascii=False, indent=2), ] return "\n".join(lines) def format_waiting_draft_troubles(result: Union[dict, list]) -> str: if not isinstance(result, list): return json.dumps(result, ensure_ascii=False, indent=2) if len(result) == 0: return "待草稿队列为空（[]）。" row = result[0] if isinstance(result[0], dict) else {} subject_id = row.get("subjectId") lines = [ "待草稿队列命中 1 条：", f"- subjectId: {subject_id}", "JSON:", json.dumps(result, ensure_ascii=False, indent=2), ] return "\n".join(lines) def format_update_work_order_draft(result: Union[dict, list]) -> str: if not isinstance(result, dict): return json.dumps(result, ensure_ascii=False, indent=2) return "工单草稿已更新:\n" + json.dumps(result, ensure_ascii=False, indent=2) def format_update_draft(result: Union[dict, list]) -> str: if not isinstance(result, dict): return json.dumps(result, ensure_ascii=False, indent=2) return "工单草稿与回复草稿已更新:\n" + json.dumps( result, ensure_ascii=False, indent=2 ) def format_mark_work_order_error(result: Union[dict, list]) -> str: if not isinstance(result, dict): return json.dumps(result, ensure_ascii=False, indent=2) return "工单状态已标记为错误:\n" + json.dumps(result, ensure_ascii=False, indent=2) def format_mark_work_order_operation_handled(result: Union[dict, list]) -> str: if not isinstance(result, dict): return json.dumps(result, ensure_ascii=False, indent=2) return "工单状态已标记为运营已处理:\n" + json.dumps( result, ensure_ascii=False, indent=2 ) def format_operation_processing_work_order_count(result: Union[dict, list]) -> str: if not isinstance(result, dict): return json.dumps(result, ensure_ascii=False, indent=2) data = result.get("data") if isinstance(result.get("data"), dict) else {} total = int(data.get("total") or 0) by_project = data.get("byProject") if isinstance(data.get("byProject"), list) else [] lines = [f"运营处理中工单总数: {total}", "按项目分布:"] if len(by_project) == 0: lines.append("- 无") else: for item in by_project: if not isinstance(item, dict): continue lines.append(f"- pid={int(item.get('pid') or 0)}: {int(item.get('count') or 0)}") lines.append("JSON:") lines.append(json.dumps(result, ensure_ascii=False, indent=2)) return "\n".join(lines) def main() -> None: parser = argparse.ArgumentParser( description="customer-service skill helper（仅保留服务端当前可用接口）" ) parser.add_argument("--json", action="store_true", help="输出原始 JSON") parser.add_argument( "--list-projects", action="store_true", help="获取项目列表（pn_data_analysis:dataType=listProject）", ) parser.add_argument( "--waiting-draft-troubles", action="store_true", help="获取待草稿客诉（skill_for_customer_service:api=getWaitingDraftTroubles）", ) parser.add_argument( "--waiting-draft-trouble-by-id", type=int, metavar="SUBJECT_ID", help="按客诉 subjectId 获取待草稿详情（skill_for_customer_service:api=getWaitingDraftTroubleById）", ) parser.add_argument( "--update-work-order-draft", type=int, metavar="SUBJECT_ID", help="更新/新建客诉工单草稿（skill_for_customer_service:api=updateWorkOrderDraft）", ) parser.add_argument( "--update-draft", type=int, metavar="SUBJECT_ID", help="同时更新工单草稿和回复草稿（skill_for_customer_service:api=updateDraft）", ) parser.add_argument( "--title", default="", help="工单标题（--update-work-order-draft / --mark-work-order-operation-handled 必填）", ) parser.add_argument( "--content", default="", help="工单内容 / 回复草稿正文（--update-work-order-draft / --mark-work-order-operation-handled / --update-reply-draft 必填）", ) parser.add_argument( "--work-order-title", default="", help="工单标题（--update-draft 必填）", ) parser.add_argument( "--work-order-content", default="", help="工单内容（--update-draft 必填）", ) parser.add_argument( "--reply-content", default="", help="回复草稿内容（--update-draft 必填）", ) parser.add_argument( "--statistic-reply", nargs=2, type=int, metavar=("START_UNIX", "END_UNIX"), help="客服回复统计（pn_data_analysis:dataType=statisticCustomerServiceReplyForSkill）", ) parser.add_argument( "--show-pn-token", nargs="?", const="", default=False, metavar="TOKEN", help="解析 token 的 uid/exp；不传则读取环境变量 PN_DATA_ANALYSIS_TOKEN", ) parser.add_argument( "--mark-work-order-error", type=int, metavar="SUBJECT_ID", help="按客诉 subjectId 将工单标记为错误（skill_for_customer_service:api=markWorkOrderError）", ) parser.add_argument( "--mark-work-order-operation-handled", type=int, metavar="SUBJECT_ID", help="按客诉 subjectId 将工单标记为运营已处理（skill_for_customer_service:api=markWorkOrderOperationHandled）", ) parser.add_argument( "--update-reply-draft", type=int, metavar="SUBJECT_ID", help="更新 AI 回复草稿（skill_for_customer_service:api=updateReplyDraft）", ) parser.add_argument( "--reply-by-draft-id", metavar="DRAFT_ID", help="采用 AI 回复草稿回复客诉（skill_for_customer_service:api=replyByDraftId）", ) parser.add_argument( "--subject-id", type=int, metavar="SUBJECT_ID", help="用于 --reply-by-draft-id：草稿所属客诉 subjectId", ) parser.add_argument( "--resolved", action="store_true", help="用于 --reply-by-draft-id：回复后标记客诉已解决", ) parser.add_argument( "--trouble-detail", type=int, metavar="SUBJECT_ID", help="按客诉 subjectId 查询详情（skill_for_customer_service:api=getTroubleDetailBySubjectId）", ) parser.add_argument( "--trouble-list-by-project", type=int, metavar="PID", help="按项目查询客诉列表（skill_for_customer_service:api=getTroubleListByProject）", ) parser.add_argument( "--trouble-list-all-projects", type=int, metavar="EID", help="查询全项目客诉列表（skill_for_customer_service:api=getTroubleListForAllProjects）", ) parser.add_argument( "--operation-processing-count", action="store_true", help="查询运营处理中工单数量（skill_for_customer_service:api=getOperationProcessingWorkOrderCount）", ) parser.add_argument( "--update-customer-service-tags", type=int, metavar="SUBJECT_ID", help="更新客诉标签（skill_for_customer_service:api=updateCustomerServiceTags）", ) parser.add_argument( "--tags-json", default="", help='用于 --update-customer-service-tags：标签数组 JSON（如 --tags-json \'["vip","bug"]\'；传 [] 清空标签）', ) parser.add_argument( "--data-json", default="", help='列表查询筛选 JSON（如 --data-json \'{"status": -1, "tags": ["充值问题"], "limit": 20, "page": 1}\'；完整有效参数见 SKILL.md / README.md）', ) parser.add_argument( "--ai-source", default="", help="用于 --update-draft / --update-reply-draft：AI 来源标识（可选）", ) args = parser.parse_args() selected = sum( 1 for enabled in [ bool(args.list_projects), bool(args.waiting_draft_troubles), args.waiting_draft_trouble_by_id is not None, args.update_work_order_draft is not None, args.update_draft is not None, args.mark_work_order_error is not None, args.mark_work_order_operation_handled is not None, args.update_reply_draft is not None, args.reply_by_draft_id is not None, args.trouble_detail is not None, args.trouble_list_by_project is not None, args.trouble_list_all_projects is not None, bool(args.operation_processing_count), args.update_customer_service_tags is not None, bool(args.statistic_reply), args.show_pn_token is not False, ] if enabled ) if selected > 1: raise SystemExit("一次仅支持一种查询模式，请只传一个子命令。") try: if args.show_pn_token is not False: tok = args.show_pn_token if args.show_pn_token else os.environ.get(ENV_TOKEN, "") if not tok: raise SystemExit( f"请传入 token：--show-pn-token '' 或配置环境变量 {ENV_TOKEN}" ) payload = decode_pn_data_analysis_token_payload(tok) out = ( json.dumps(payload, ensure_ascii=False, indent=2) if args.json else format_pn_token_payload(payload) ) elif args.list_projects: result = query_project_list() out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_project_list_result(result) ) elif args.waiting_draft_troubles: result = query_waiting_draft_troubles() out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_waiting_draft_troubles(result) ) elif args.waiting_draft_trouble_by_id is not None: result = query_waiting_draft_trouble_by_id(args.waiting_draft_trouble_by_id) out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_waiting_draft_troubles(result) ) elif args.update_work_order_draft is not None: title = (args.title or "").strip() content = (args.content or "").strip() if title == "" or content == "": raise SystemExit("--update-work-order-draft 需要同时传 --title 和 --content") result = query_update_work_order_draft( args.update_work_order_draft, title, content, ) out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_update_work_order_draft(result) ) elif args.update_draft is not None: work_order_title = (args.work_order_title or "").strip() work_order_content = (args.work_order_content or "").strip() reply_content = (args.reply_content or "").strip() if work_order_title == "" or work_order_content == "" or reply_content == "": raise SystemExit( "--update-draft 需要同时传 --work-order-title、--work-order-content 和 --reply-content" ) result = query_update_draft( args.update_draft, work_order_title, work_order_content, reply_content, args.ai_source, ) out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_update_draft(result) ) elif args.mark_work_order_error is not None: result = query_mark_work_order_error( args.mark_work_order_error, ) out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_mark_work_order_error(result) ) elif args.mark_work_order_operation_handled is not None: title = (args.title or "").strip() content = (args.content or "").strip() if title == "" or content == "": raise SystemExit( "--mark-work-order-operation-handled 需要同时传 --title 和 --content" ) result = query_mark_work_order_operation_handled( args.mark_work_order_operation_handled, title, content, ) out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_mark_work_order_operation_handled(result) ) elif args.update_reply_draft is not None: content = (args.content or "").strip() if content == "": raise SystemExit("--update-reply-draft 需要传 --content") result = query_update_reply_draft( args.update_reply_draft, content, args.ai_source, ) out = json.dumps(result, ensure_ascii=False, indent=2) elif args.reply_by_draft_id is not None: draft_id = (args.reply_by_draft_id or "").strip() if draft_id == "" or args.subject_id is None or args.subject_id <= 0: raise SystemExit( "--reply-by-draft-id 需要同时传 --subject-id " ) result = query_reply_by_draft_id( args.subject_id, draft_id, bool(args.resolved), ) out = json.dumps(result, ensure_ascii=False, indent=2) elif args.trouble_detail is not None: result = query_trouble_detail_by_subject_id(args.trouble_detail) out = json.dumps(result, ensure_ascii=False, indent=2) elif args.trouble_list_by_project is not None: data = _parse_json_object(args.data_json, "--data-json") result = query_trouble_list_by_project(args.trouble_list_by_project, data) out = json.dumps(result, ensure_ascii=False, indent=2) elif args.trouble_list_all_projects is not None: data = _parse_json_object(args.data_json, "--data-json") result = query_trouble_list_for_all_projects( args.trouble_list_all_projects, data ) out = json.dumps(result, ensure_ascii=False, indent=2) elif args.operation_processing_count: result = query_operation_processing_work_order_count() out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_operation_processing_work_order_count(result) ) elif args.update_customer_service_tags is not None: tags_data = args.tags_json.strip() if tags_data == "": raise SystemExit("--update-customer-service-tags 需要传 --tags-json") try: parsed_tags = json.loads(tags_data) except json.JSONDecodeError as e: raise SystemExit(f"--tags-json 不是合法 JSON: {e.msg}") from e if not isinstance(parsed_tags, list): raise SystemExit("--tags-json 必须是 JSON 数组，例如 '[\"vip\",\"bug\"]'") tags = [] for tag in parsed_tags: if not isinstance(tag, str): raise SystemExit("--tags-json 必须是字符串数组") text = tag.strip() if text != "": tags.append(text) if len(tags) > 3: raise SystemExit("--tags-json 最多支持 3 个标签") result = query_update_customer_service_tags( args.update_customer_service_tags, tags ) out = json.dumps(result, ensure_ascii=False, indent=2) elif args.statistic_reply: start_ts, end_ts = args.statistic_reply result = query_statistic_customer_service_reply_for_skill( start_ts, end_ts ) out = ( json.dumps(result, ensure_ascii=False, indent=2) if args.json else format_statistic_customer_service_reply_for_skill_result(result) ) else: parser.print_help() return except Exception as e: print(f"请求失败: {e}", file=sys.stderr) sys.exit(1) print(out) if __name__ == "__main__": main()