import requests import json import datetime import time # 用于时间转换 import gradio as gr # --- 抖音配置信息 --- DOUYIN_CLIENT_KEY = 'awbeykzyos7kbidv' DOUYIN_CLIENT_SECRET = '4575440b156ecbe144284e4f69d284a2' # Keep secure! DOUYIN_ACCOUNT_ID = '7241078611527075855' DOUYIN_TOKEN_URL = 'https://open.douyin.com/oauth/client_token/' DOUYIN_ORDER_QUERY_URL = 'https://open.douyin.com/goodlife/v1/trade/order/query/' # --- 飞书配置信息 --- FEISHU_APP_ID = "cli_a6672cae343ad00e" FEISHU_APP_SECRET = "0J4SpfBMeIxJEOXDJMNbofMipRgwkMpV" # Keep secure! FEISHU_APP_TOKEN = "GcTMbpdDPaAxV5sb8gvcRVC3ndh" # 表格 1 (订单记录) FEISHU_TABLE1_ID = "tblaYUUrFT1rKXNB" FEISHU_TABLE1_ORDER_FIELD = "订单号" # 文本类型 FEISHU_TABLE1_STATUS_FIELD = "履约状态" # 文本类型 - 将被更新为 "已兑换" TABLE1_EXCHANGED_STATUS_CONTENT = "已兑换" # Table 1 更新后的状态 # 表格 2 (激活码/发送记录) FEISHU_TABLE2_ID = "tbl0S1kl9FTBX8oT" FEISHU_TABLE2_ACTIVATION_CODE_FIELD = "激活码" # Table 2 中激活码字段名 (假设为文本) FEISHU_TABLE2_STATUS_FIELD = "发送状态" # Table 2 中发送状态字段名 (文本类型) FEISHU_TABLE2_SORT_FIELD = "最后更新时间" # Table 2 中用于排序的字段名 FEISHU_TABLE2_UPDATE_TIME_FIELD = "最后更新时间" # Table 2 中最后更新日期字段名 TABLE2_UNSENT_STATUS_CONTENT = "未发送" # Table 2 中查找的状态 TABLE2_SENT_STATUS_CONTENT = "已发送" # Table 2 更新后的状态 # --- 抖音券（item）状态映射字典 --- ITEM_STATUS_MAP = { 0: "初始化", 1: "交易成功", 10: "待支付", 20: "支付成功", 100: "待使用", 101: "交易关闭", 200: "预约中", 201: "已预约", 300: "退款中", 301: "已退款", 400: "履约中", 401: "已履约" } # =========== 必须包含所有 API 函数定义 =========== def get_douyin_access_token(client_key, client_secret): headers = {'Content-Type': 'application/json'} payload = {"grant_type": "client_credential", "client_key": client_key, "client_secret": client_secret} try: response = requests.post(DOUYIN_TOKEN_URL, headers=headers, json=payload, timeout=10); response.raise_for_status(); data = response.json() if data.get('data') and data['data'].get('error_code') == 0: return data['data'].get('access_token'), "成功获取抖音 Access Token！" else: msg = f"获取抖音 Access Token 失败: Code={data.get('data', {}).get('error_code', 'N/A')}, Desc={data.get('data', {}).get('description', '未知错误')}"; return None, msg except Exception as e: msg = f"请求抖音 Access Token 时发生错误: {e}"; return None, msg def query_douyin_order(access_token, account_id, order_id_str): if not access_token: return None, "抖音 Token 无效" headers = {'Content-Type': 'application/json', 'access-token': access_token} params = {'account_id': account_id, 'order_id': order_id_str, 'page_num': 1, 'page_size': 1} try: response = requests.get(DOUYIN_ORDER_QUERY_URL, headers=headers, params=params, timeout=15); response.raise_for_status(); data = response.json() extra_error_code = data.get('extra', {}).get('error_code', 0); data_error_code = data.get('data', {}).get('error_code', 0) if extra_error_code == 0 and data_error_code == 0: return data, f"抖音订单 '{order_id_str}' 查询成功！" else: error_code = extra_error_code if extra_error_code != 0 else data_error_code; description = data.get('extra', {}).get('description', '') or data.get('data', {}).get('description', '未知错误'); msg = f"抖音订单查询失败: Code={error_code}, Desc={description}"; return None, msg except Exception as e: msg = f"查询抖音订单时发生错误: {e}"; return None, msg def parse_douyin_order_status(order_data): try: if not order_data or 'data' not in order_data or 'orders' not in order_data['data'] or not order_data['data']['orders']: return None, None, "抖音订单数据无效。" order = order_data['data']['orders'][0]; certificates = order.get('certificate', []) if not certificates: return None, None, "订单中未找到券信息。" first_cert_status_code = certificates[0].get('item_status'); certificate_id = certificates[0].get('certificate_id', 'N/A') if first_cert_status_code is not None: status_text = ITEM_STATUS_MAP.get(first_cert_status_code, f"未知状态码({first_cert_status_code})"); msg=f"解析到抖音订单状态: {status_text} (状态码: {first_cert_status_code}, 券ID: {certificate_id})"; return status_text, first_cert_status_code, msg else: return None, None, "第一个券信息中未找到 'item_status'。" except Exception as e: msg = f"解析抖音订单状态时出错: {e}"; return None, None, msg def get_feishu_tenant_access_token(app_id, app_secret): auth_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" payload = json.dumps({"app_id": app_id, "app_secret": app_secret}) headers = {'Content-Type': 'application/json; charset=utf-8'} try: response = requests.post(auth_url, headers=headers, data=payload, timeout=10); response.raise_for_status(); data = response.json() if data.get("code") == 0: return data.get("tenant_access_token"), "获取飞书 tenant_access_token 成功！" else: msg = f"获取飞书 tenant_access_token 失败: {data.get('msg')} (Code: {data.get('code')})"; return None, msg except Exception as e: msg = f"请求飞书 tenant_access_token 时发生错误: {e}"; return None, msg def check_order_exists_in_feishu(access_token, app_token, table_id, order_field_name, order_id_str_to_check): if not access_token: return None, "飞书 Access Token 无效，无法检查记录。" search_url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search" headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json; charset=utf-8'} filter_payload = {"filter": {"conjunction": "and", "conditions": [{"field_name": order_field_name, "operator": "is", "value": [ order_id_str_to_check ]}]}, "page_size": 1} msg = f"正在 Table 1 ({table_id}) 中检查订单号 '{order_id_str_to_check}'..." try: response = requests.post(search_url, headers=headers, json=filter_payload, timeout=15); response.raise_for_status(); data = response.json() if data.get("code") == 0: items = data.get("data", {}).get("items", []) exists = bool(items) msg += f" 检查结果: {'已存在' if exists else '不存在'}。" return exists, msg # 返回布尔值和消息 else: msg += f" 检查失败: Code={data.get('code')}, Msg={data.get('msg')}" return None, msg # 返回 None 表示检查出错 except Exception as e: msg += f" 检查时发生错误: {e}" return None, msg def add_bitable_record(access_token, app_token, table_id, order_field_name, order_id_str, status_field_name, exchange_status): if not access_token: return None, "飞书 Access Token 无效，无法写入数据。" write_url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records" headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json; charset=utf-8'} payload = {"fields": {order_field_name: order_id_str, status_field_name: exchange_status}} msg = f"准备写入 Table 1 ({table_id})..." try: response = requests.post(write_url, headers=headers, json=payload, timeout=15); response.raise_for_status(); data = response.json() if data.get("code") == 0: record_info = data.get('data', {}).get('record') record_id = record_info.get('record_id') if record_info else None msg += f" Table 1 写入成功！Record ID: {record_id}" return record_id, msg # 返回 record_id 和消息 else: msg += f" Table 1 写入失败: Code={data.get('code')}, Msg={data.get('msg')}" return None, msg except Exception as e: msg += f" Table 1 写入时发生错误: {e}" return None, msg # --- 修改：查找 Table 2 记录，并只提取激活码文本 --- def find_first_unsent_record_and_code_text(access_token, app_token, table_id, status_field, unsent_status_value, activation_code_field, sort_field): """ 在 Table 2 查找第一个状态为 '未发送' 的记录，返回其 record_id 和激活码文本。 """ if not access_token: print("内部错误: 飞书 Token 无效"); return None, None search_url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search" headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json; charset=utf-8'} payload = { "filter": {"conjunction": "and", "conditions": [{"field_name": status_field, "operator": "is", "value": [unsent_status_value]}]}, "sort": [{"field_name": sort_field, "order": "asc"}], "page_size": 1, # 【修改这里】: 直接传递 Python 列表，不要 json.dumps() "field_names": [activation_code_field] } print(f"飞书API: 查找 Table 2 '{status_field}' 为 '{unsent_status_value}' 的记录...") # 后台日志 # print("构造的查找 Payload:", json.dumps(payload, indent=2, ensure_ascii=False)) # 调试时可以打印看看 try: response = requests.post(search_url, headers=headers, json=payload, timeout=15); response.raise_for_status(); data = response.json() if data.get("code") == 0: items = data.get("data", {}).get("items", []) if items: # ... (后续解析激活码的逻辑不变) ... record = items[0]; record_id = record.get('record_id') activation_code_raw = record.get('fields', {}).get(activation_code_field) activation_code_text = None if isinstance(activation_code_raw, list) and len(activation_code_raw) > 0 and isinstance(activation_code_raw[0], dict) and 'text' in activation_code_raw[0]: activation_code_text = activation_code_raw[0]['text'] elif isinstance(activation_code_raw, str): activation_code_text = activation_code_raw elif activation_code_raw is not None: try: activation_code_text = str(activation_code_raw); print(f"警告: 激活码字段值类型为 {type(activation_code_raw)}, 已尝试转为字符串。") except: print(f"警告: 无法将激活码字段值 {activation_code_raw} 转为字符串。") else: print(f"警告: 激活码字段值为空或格式未知: {activation_code_raw}") if record_id and activation_code_text is not None: print(f"飞书API: 找到记录 ID: {record_id}, 激活码文本: {activation_code_text}"); return record_id, activation_code_text elif record_id: print(f"飞书API: 找到记录 ID: {record_id}, 但未能提取激活码文本。"); return record_id, None else: return None, None else: print(f"飞书API: 未找到 '{status_field}' 为 '{unsent_status_value}' 的记录。"); return None, None else: print(f"飞书API: 查找 Table 2 记录失败: Code={data.get('code')}, Msg={data.get('msg')}"); return None, None except Exception as e: print(f"飞书API: 查找 Table 2 记录时发生错误: {e}"); return None, None def update_table2_record_to_sent(access_token, app_token, table_id, record_id, status_field, sent_status_content, update_time_field): if not access_token or not record_id: return False, "飞书 Token 或 Table 2 Record ID 无效，无法更新。" update_url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_update" headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json; charset=utf-8'} now = datetime.datetime.now(); formatted_time_str = now.strftime("%Y/%m/%d %H:%M") payload_fields = {status_field: sent_status_content, update_time_field: formatted_time_str} payload = {"records": [{"record_id": record_id, "fields": payload_fields}]} msg = f"准备更新 Table 2 ({table_id}, ID: {record_id}) 为 '{sent_status_content}'..." try: response = requests.post(update_url, headers=headers, json=payload, timeout=15); response.raise_for_status(); data = response.json() if data.get("code") == 0: failed_records = data.get("data", {}).get("failed_records", []) if not failed_records: msg += " Table 2 更新成功！"; return True, msg else: msg += f" Table 2 更新失败: {failed_records}"; return False, msg else: msg += f" Table 2 更新失败: Code={data.get('code')}, Msg={data.get('msg')}"; return False, msg except Exception as e: msg += f" Table 2 更新时发生错误: {e}"; return False, msg def update_table1_record_to_exchanged(access_token, app_token, table_id, record_id, status_field, exchanged_status_content): if not access_token or not record_id: return False, "飞书 Token 或 Table 1 Record ID 无效，无法更新。" update_url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_update" headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json; charset=utf-8'} payload = {"records": [{"record_id": record_id, "fields": {status_field: exchanged_status_content}}]} msg = f"准备更新 Table 1 ({table_id}, ID: {record_id}) 为 '{exchanged_status_content}'..." try: response = requests.post(update_url, headers=headers, json=payload, timeout=15); response.raise_for_status(); data = response.json() if data.get("code") == 0: failed_records = data.get("data", {}).get("failed_records", []) if not failed_records: msg += " Table 1 更新成功！"; return True, msg else: msg += f" Table 1 更新失败: {failed_records}"; return False, msg else: msg += f" Table 1 更新失败: Code={data.get('code')}, Msg={data.get('msg')}"; return False, msg except Exception as e: msg += f" Table 1 更新时发生错误: {e}"; return False, msg # --- Gradio 主处理函数 (简化版) --- def process_order_simple(douyin_order_id_str): final_output = "处理中..." # --- 1. 获取抖音 Token --- douyin_token, msg = get_douyin_access_token(DOUYIN_CLIENT_KEY, DOUYIN_CLIENT_SECRET) if not douyin_token: return f"错误(抖音Token): {msg}" # --- 2. 验证输入 --- douyin_order_id_str = douyin_order_id_str.strip() if not douyin_order_id_str: return "请输入抖音订单号" # --- 3. 查询抖音订单 --- order_details, msg = query_douyin_order(douyin_token, DOUYIN_ACCOUNT_ID, douyin_order_id_str) if not order_details: return f"错误(查询订单): {msg}" # --- 4. 解析抖音状态 --- fulfillment_status_text, _, msg = parse_douyin_order_status(order_details) if fulfillment_status_text is None: return f"错误(解析状态): {msg}" # --- 5. 检查抖音状态 --- if fulfillment_status_text != "已履约": return f"提示：订单状态为 '{fulfillment_status_text}'，不予兑换。" # --- 6. 获取飞书 Token --- feishu_token, msg = get_feishu_tenant_access_token(FEISHU_APP_ID, FEISHU_APP_SECRET) if not feishu_token: return f"错误(飞书Token): {msg}" # --- 7. 检查 Table 1 --- exists_in_table1, msg = check_order_exists_in_feishu(feishu_token, FEISHU_APP_TOKEN, FEISHU_TABLE1_ID, FEISHU_TABLE1_ORDER_FIELD, douyin_order_id_str) if exists_in_table1 is None: return f"错误(检查Table1): {msg}" if exists_in_table1 is True: return f"提示：订单 '{douyin_order_id_str}' 已兑换过。" # --- 8. 写入 Table 1 --- table1_record_id, msg = add_bitable_record(feishu_token, FEISHU_APP_TOKEN, FEISHU_TABLE1_ID, FEISHU_TABLE1_ORDER_FIELD, douyin_order_id_str, FEISHU_TABLE1_STATUS_FIELD, fulfillment_status_text) if not table1_record_id: return f"错误(写入Table1): {msg}" print(f"后台日志: Table 1 写入成功, Record ID: {table1_record_id}") # --- 9. 查找 Table 2 --- table2_record_id, activation_code_text = find_first_unsent_record_and_code_text(feishu_token, FEISHU_APP_TOKEN, FEISHU_TABLE2_ID, FEISHU_TABLE2_STATUS_FIELD, TABLE2_UNSENT_STATUS_CONTENT, FEISHU_TABLE2_ACTIVATION_CODE_FIELD, FEISHU_TABLE2_SORT_FIELD) if table2_record_id and activation_code_text: final_output = f"{activation_code_text}" # 直接显示激活码 # --- 10. "发送" 激活码 (模拟) --- print(f"后台日志: '发送' 激活码: {activation_code_text}") sent_success = True # 假设发送成功 if sent_success: # --- 11. 更新 Table 2 --- update_t2_success, msg_t2 = update_table2_record_to_sent(feishu_token, FEISHU_APP_TOKEN, FEISHU_TABLE2_ID, table2_record_id, FEISHU_TABLE2_STATUS_FIELD, TABLE2_SENT_STATUS_CONTENT, FEISHU_TABLE2_UPDATE_TIME_FIELD) if not update_t2_success: print(f"后台警告: {msg_t2}") # 只在后台打印警告 if update_t2_success: # 只有 T2 更新成功才更新 T1 # --- 12. 更新 Table 1 --- update_t1_success, msg_t1 = update_table1_record_to_exchanged(feishu_token, FEISHU_APP_TOKEN, FEISHU_TABLE1_ID, table1_record_id, FEISHU_TABLE1_STATUS_FIELD, TABLE1_EXCHANGED_STATUS_CONTENT) if not update_t1_success: print(f"后台警告: {msg_t1}") # 只在后台打印警告 else: final_output = "错误：激活码 '发送' 失败 (模拟)" elif table2_record_id and not activation_code_text: final_output = f"错误：未能获取激活码文本，请检查 Table 2 数据。" else: final_output = "提示：无可用激活码，请联系管理员。" return final_output # 返回最终结果 # --- 创建 Gradio Interface (最终简化版) --- with gr.Blocks(theme=gr.themes.Soft()) as demo: gr.Markdown("# 竞潮玩游戏激活码兑换小助手") gr.Markdown("请输入抖音订单号以兑换激活码。仅限状态为“已履约”且未兑换过的订单。") with gr.Row(): order_id_input = gr.Textbox(label="输入抖音订单号", placeholder="在此填写订单号...") process_button = gr.Button("兑换激活码") with gr.Row(): activation_code_output = gr.Textbox(label="兑换结果 / 激活码", interactive=False) process_button.click( fn=process_order_simple, inputs=order_id_input, outputs=activation_code_output ) # --- 启动 Gradio 应用 --- if __name__ == "__main__": demo.launch()