import json import re import urllib.parse import urllib.request import urllib.error import pymssql from datetime import datetime, timedelta from qcloud_cos import CosConfig from qcloud_cos import CosS3Client # API流程文件fileUrl为OSS相对路径，拼接基础地址得到完整下载URL（与config.php source_oss_base_url 一致） DEFAULT_OSS_BASE_URL = 'https://usaeu-1259285998.cos.ap-guangzhou.myqcloud.com' def resolve_file_url(file_url): """ 将文件URL解析为可下载的完整URL - 已以 http(s):// 开头的完整URL 原样返回 - 相对路径（OSS 相对路径）拼接文件级常量 DEFAULT_OSS_BASE_URL 参数: file_url: 文件URL或相对路径 返回: str: 完整URL；file_url为空时返回空字符串 """ file_url = (file_url or '').strip() if not file_url: return '' if re.match(r'^https?://', file_url, re.IGNORECASE): return file_url base = DEFAULT_OSS_BASE_URL.rstrip('/') return base + '/' + file_url.lstrip('/') def parse_legal_signed_file_json(value): """ 解析 API 流程的法人签名文件值（[{fileUrl,fileName}] JSON 数组字符串） fileUrl 为 OSS 相对路径或完整URL，解析为可下载的完整URL 参数: value: JSON 数组字符串，如 [{"fileUrl":"common-test/xxx.png","fileName":"xxx.png"}] 返回: str: 完整URL；解析失败或缺少fileUrl时返回空字符串 """ try: arr = json.loads(value) except (ValueError, TypeError): return '' if not isinstance(arr, list) or not arr: return '' first = arr[0] if not isinstance(first, dict): return '' file_url = first.get('fileUrl') or first.get('FileUrl') or '' return resolve_file_url(file_url) def build_cos_client(secret_id, secret_key, region): """ 构建 COS 客户端（签名用） 参数: secret_id/secret_key/region: 腾讯云凭证与地域 返回: CosS3Client 或 None（凭证/地域缺失或构建失败时返回 None，由调用方走无签名兜底） """ if not (secret_id and secret_key and region): return None try: return CosS3Client(CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key)) except Exception as e: print(f"构建COS客户端失败: {e}") return None def build_signed_cos_url(client, bucket, region, url_or_key, expire_seconds=600): """ 生成 COS 带签名下载URL（API 流程 usaeu 私有桶对象下载需带验证信息，仿 PHP CosDownloader::signBucketUrl） - 相对路径：固定域名 + key 拼接后生成 q-sign-* 签名 - 完整 URL 命中 COS 桶域名且未带签名：按对象键重新生成签名（获取执行时刻的新鲜签名） - 完整 URL 已带签名 / 本地路径 / 其他域名：原样返回 - 签名生成失败：原样返回（由调用方按无签名直连兜底，公读对象仍可用） - source 流程（vat 公读桶/文件服务器域名）不走本函数，无需签名 参数: client: CosS3Client bucket: COS 桶名 region: COS 地域,如 'ap-guangzhou' url_or_key: 完整URL或COS相对路径 expire_seconds: 签名有效期（秒），默认 600（10分钟） 返回: str: 带签名下载URL；入参为空时返回空串 """ url_or_key = (url_or_key or '').strip() if not url_or_key: return '' domain = f'{bucket}.cos.{region}.myqcloud.com' if re.match(r'^https?://', url_or_key, re.IGNORECASE): parsed = urllib.parse.urlparse(url_or_key) if parsed.netloc.lower() != domain.lower(): return url_or_key if 'sign=' in parsed.query or 'q-sign-' in parsed.query: return url_or_key key = urllib.parse.unquote(parsed.path.lstrip('/')) else: key = url_or_key.lstrip('/') try: signed = client.get_presigned_url(Bucket=bucket, Key=key, Expired=expire_seconds, Method='get') if signed: return signed except Exception as e: print(f"生成签名URL失败（兜底无签名直连）: {e}") return url_or_key def sign_cos_url(secret_id, secret_key, bucket, region, url_or_key, expire_seconds=600): """ 按分开传入的 COS 凭证（secret_id/secret_key/bucket/region）生成带签名下载URL - 内部组合 build_cos_client + build_signed_cos_url - 凭证/地域任一为空时不签名，返回原URL（未传签名能力时公读直连兜底） - 签名失败同样返回原URL（下游按无签名直连兜底） 参数: secret_id: 腾讯云 SecretId secret_key: 腾讯云 SecretKey bucket: COS 桶名 region: COS 地域,如 'ap-guangzhou' url_or_key: 完整URL或COS相对路径 expire_seconds: 签名有效期（秒），默认 600（10分钟） 返回: str: 带签名下载URL；无法签名时原样返回 url_or_key """ if not (secret_id and secret_key and bucket and region): return url_or_key client = build_cos_client(secret_id, secret_key, region) if not client: return url_or_key return build_signed_cos_url(client, bucket, region, url_or_key, expire_seconds) def probe_cos_object(url, timeout=20): """ GET+Range(bytes=0-0) 探测 COS 对象是否可下载（下载前/下载失败后判断对象是否有效） - 陷阱：COS 签名 URL 绑定 GET 方法——对签名 URL 发 HEAD 请求必 403（获取文件信息失败）， 探测存在性/大小只能用 GET（配合 Range: bytes=0-0 只取 1 字节，200/206 即对象存在； Content-Range 头可读到对象总大小） - 403 时输出诊断：签名过期（600 秒）→ 需重新调用取数流程获取新签名再试；私有桶无签名直连也会 403 参数: url: COS 对象完整 URL（签名或未签名均可） timeout: 探测超时秒数 返回: bool: 200/206 返回 True，403/超时/网络错误返回 False """ try: req = urllib.request.Request(url, headers={'Range': 'bytes=0-0'}) with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status in (200, 206) except urllib.error.HTTPError as e: if e.code == 403: print( f"探测COS对象被拒绝(403): {url[:100]}... " f"[可能原因] 签名URL过期(600秒)请重新取数获取新签名；" f"或对象为私有桶且URL无签名（GET+Range 也需带签名）" ) else: print(f"探测COS对象失败: HTTP {e.code} {url[:100]}...") return False except Exception as e: print(f"探测COS对象出错: {e} {url[:100]}...") return False def process_datetime_fields(result): """ 处理结果中的datetime字段，将datetime对象转换为字符串 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ for key, value in result.items(): if isinstance(value, datetime): result[key] = value.strftime('%Y-%m-%d %H:%M:%S') def process_legal_person_birth_date(result): """ 处理LegalPersonBirthDate字段，分解为年月日，删除原字段 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonBirthDate'): birth_date_str = result['LegalPersonBirthDate'] try: # 尝试解析日期字符串 birth_date = datetime.strptime(birth_date_str, '%Y-%m-%d %H:%M:%S') result['LegalPersonBirthDate_Y'] = birth_date.year result['LegalPersonBirthDate_M'] = birth_date.month result['LegalPersonBirthDate_D'] = birth_date.day except ValueError: # 如果解析失败，尝试其他可能的格式 try: birth_date = datetime.strptime(birth_date_str[:19], '%Y-%m-%d %H:%M:%S') result['LegalPersonBirthDate_Y'] = birth_date.year result['LegalPersonBirthDate_M'] = birth_date.month result['LegalPersonBirthDate_D'] = birth_date.day except ValueError: # 如果还是失败，设为默认值 result['LegalPersonBirthDate_Y'] = '' result['LegalPersonBirthDate_M'] = '' result['LegalPersonBirthDate_D'] = '' # 删除原字段 del result['LegalPersonBirthDate'] def process_legal_person_pinyin_name(result): """ 处理LegalPersonFullNamePinYin字段，分解为姓和名，删除原字段 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonFullNamePinYin'): pinyin_name = result['LegalPersonFullNamePinYin'].strip().upper() parts = pinyin_name.split() if len(parts) >= 2: # 最后一个部分为姓，前面的为名 result['LegalPersonFullNamePinYin_X'] = parts[-1] result['LegalPersonFullNamePinYin_M'] = ''.join(parts[:-1]) elif len(parts) == 1: # 只有一个部分时，作为姓处理 result['LegalPersonFullNamePinYin_X'] = parts[0] result['LegalPersonFullNamePinYin_M'] = '' else: # 空字符串处理 result['LegalPersonFullNamePinYin_X'] = '' result['LegalPersonFullNamePinYin_M'] = '' # 删除原字段 del result['LegalPersonFullNamePinYin'] def process_legal_person_city_english_name(result): """ 处理LegalPersonCityEngName字段，将字符全部变为大写，去除中间的空格，去掉最后的shi拼音 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonCityEngName'): city_name = result['LegalPersonCityEngName'].strip() # 转换为大写 city_name = city_name.upper() # 去除中间的空格 city_name = city_name.replace(' ', '') # 去掉最后的shi拼音 if city_name.endswith('SHI'): city_name = city_name[:-3] # 直接在原字段上重新赋值 result['LegalPersonCityEngName'] = city_name def process_legal_person_address_province_english_name(result): """ 处理LegalPersonAddressProvinceEn字段，将字符全部变为大写，去除中间的空格，去掉最后的sheng拼音 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonAddressProvinceEn'): province_name = result['LegalPersonAddressProvinceEn'].strip() # 转换为大写 province_name = province_name.upper() # 去除中间的空格 province_name = province_name.replace(' ', '') # 去掉最后的sheng拼音 if province_name.endswith('SHENG'): province_name = province_name[:-5] # 直接在原字段上重新赋值 result['LegalPersonAddressProvinceEn'] = province_name def process_legal_person_country(result): """ 处理LegalPersonCountry字段，将字符全部变为大写 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('LegalPersonCountry'): # 直接在原字段上重新赋值 result['LegalPersonCountry'] = result['LegalPersonCountry'].strip().upper() if result.get('CompanyCountry'): # 直接在原字段上重新赋值 result['CompanyCountry'] = result['CompanyCountry'].strip().upper() def process_city_english_name(result): """ 处理CityEngName字段，将字符全部变为大写，去除中间的空格，去掉最后的shi拼音 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('CityEngName'): city_name = result['CityEngName'].strip() # 转换为大写 city_name = city_name.upper() # 去除中间的空格 city_name = city_name.replace(' ', '') # 去掉最后的shi拼音 if city_name.endswith('SHI'): city_name = city_name[:-3] # 直接在原字段上重新赋值 result['CityEngName'] = city_name def process_company_address_province_english_name(result): """ 处理CompanyAddressProvinceEn字段，将字符全部变为大写，去除中间的空格，去掉最后的sheng拼音 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ if result.get('CompanyAddressProvinceEn'): province_name = result['CompanyAddressProvinceEn'].strip() # 转换为大写 province_name = province_name.upper() # 去除中间的空格 province_name = province_name.replace(' ', '') # 去掉最后的sheng拼音 if province_name.endswith('SHENG'): province_name = province_name[:-5] # 直接在原字段上重新赋值 result['CompanyAddressProvinceEn'] = province_name def process_id_number_and_passport(result): """ 处理身份证号和护照号码字段 LegalPersonIDCardType为IDCard时，LegalPersonIdNumber值不变，LegalPersonPassportNumber为空 LegalPersonIDCardType为其他值时，LegalPersonPassportNumber为LegalPersonIdNumber的值，LegalPersonIdNumber的值为空 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ id_card_type = result.get('LegalPersonIDCardType', '').strip() id_number = result.get('LegalPersonIdNumber', '') if id_card_type == 'IDCard': # 身份证类型，保持身份证号不变，护照号为空 result['LegalPersonIdNumber'] = id_number result['LegalPersonPassportNumber'] = '' else: # 其他类型，将身份证号的值移到护照号，身份证号设为空 result['LegalPersonPassportNumber'] = id_number result['LegalPersonIdNumber'] = '' def process_current_date(result): """ 处理当前日期，计算当天的年月日并保存到各个字段 参数: result: 查询结果字典 返回: None: 直接修改原字典 """ current_date = datetime.now() result['current_y'] = current_date.year result['current_m'] = current_date.month result['current_d'] = current_date.day def process_legal_signed_file(result, ip, port, username, password, secret_id=None, secret_key=None, bucket=None, region=None): """ 处理LegalSignedFile字段 - API流程：值为 [{fileUrl,fileName}] JSON 数组字符串（fileUrl 为 OSS 相对路径或完整URL） → 解析出可下载的完整URL（相对路径拼接文件级常量 DEFAULT_OSS_BASE_URL，与 PHP resolveFileUrl 行为一致）； 传入 COS 凭证（secret_id/secret_key/bucket/region）时对 usaeu 私有桶对象生成带签名URL（600 秒，下载需带验证信息） - 兼容：值本身为完整 http(s) URL 时原样保留 - source流程（原逻辑）：值为 Base_AnnexesFile.F_Id，连接vat_db库查询文件路径并替换字符串 参数: result: 查询结果字典 ip: 数据库IP地址 port: 数据库端口 username: 数据库用户名 password: 数据库密码 secret_id: 腾讯云 SecretId（API流程签名URL用；为空时不签名） secret_key: 腾讯云 SecretKey bucket: COS 桶名（API流程 usaeu 私有桶） region: COS 地域,如 'ap-guangzhou' 返回: None: 直接修改原字典 """ raw_value = result.get('LegalSignedFile') if not raw_value: # 如果字段不存在，添加空字符串 result['LegalSignedFile'] = '' return raw_value = str(raw_value).strip() # API流程：[{fileUrl,fileName}] JSON 数组字符串 if raw_value.startswith('['): resolved = parse_legal_signed_file_json(raw_value) # API 流程：usaeu 私有桶对象下载需带验证信息（签名 URL，600 秒）； # 未传 COS 凭证或签名失败时兜底无签名直连（公读对象仍可用） if resolved: resolved = sign_cos_url(secret_id, secret_key, bucket, region, resolved) result['LegalSignedFile'] = resolved return # 兼容：完整URL直接保留 if re.match(r'^https?://', raw_value, re.IGNORECASE): result['LegalSignedFile'] = raw_value return # source流程（原逻辑）：值为 Base_AnnexesFile.F_Id，查库取文件路径 conn = None cursor = None try: # 连接到vat_db数据库 conn = pymssql.connect( server=ip, port=port, user=username, password=password, database='vat_db' ) cursor = conn.cursor() # 查询文件路径 query_sql = """ SELECT F_FilePath FROM Base_AnnexesFile WHERE F_Id = %s """ cursor.execute(query_sql, (raw_value,)) file_result = cursor.fetchone() if file_result and file_result[0]: file_path = file_result[0] # 替换字符串路径 processed_path = file_path.replace('G:/fileAnnexes', 'https://file.usaeu.com') # 直接在原字段上重新赋值，删除原字段值 result['LegalSignedFile'] = processed_path except Exception as e: print(f"处理LegalSignedFile时发生错误: {e}") # 出错时设置为空字符串 result['LegalSignedFile'] = '' finally: # 关闭游标和连接 if cursor: cursor.close() if conn: conn.close() def process_result_fields(result, ip=None, port=None, username=None, password=None, secret_id=None, secret_key=None, bucket=None, region=None): """ 处理查询结果中的各种字段 参数: result: 查询结果字典 ip: 数据库IP地址 (用于LegalSignedFile处理) port: 数据库端口 (用于LegalSignedFile处理) username: 数据库用户名 (用于LegalSignedFile处理) password: 数据库密码 (用于LegalSignedFile处理) secret_id: 腾讯云 SecretId（API流程签名URL用；为空时不签名） secret_key: 腾讯云 SecretKey bucket: COS 桶名（API流程 usaeu 私有桶） region: COS 地域,如 'ap-guangzhou' 返回: None: 直接修改原字典 """ process_datetime_fields(result) process_legal_person_birth_date(result) # 分解为年月日，删除原字段 process_legal_person_pinyin_name(result) # 分解为姓和名，删除原字段 process_legal_person_city_english_name(result) # 直接修改原字段 process_legal_person_address_province_english_name(result) # 直接修改原字段 process_legal_person_country(result) # 直接修改原字段 process_city_english_name(result) # 直接修改原字段 process_company_address_province_english_name(result) # 直接修改原字段 process_id_number_and_passport(result) # 处理身份证号和护照号 process_current_date(result) # 新增当前日期字段 # 处理LegalSignedFile，需要数据库连接参数 if all([ip, port, username, password]): process_legal_signed_file(result, ip, port, username, password, secret_id, secret_key, bucket, region) # 直接修改原字段 else: result['LegalSignedFile'] = '' # 可以在这里添加更多字段处理函数 def query_and_update_vat_task(ip, port, username, password, database, secret_id=None, secret_key=None, bucket=None, region=None): """ 查询AsyncVATTasks表中符合条件的一条记录，并更新其状态 参数: ip: 数据库IP地址 port: 数据库端口 username: 数据库用户名 password: 数据库密码 database: 数据库名称 secret_id: 腾讯云 SecretId（API流程签名URL用；为空时不签名） secret_key: 腾讯云 SecretKey bucket: COS 桶名（API流程 usaeu 私有桶） region: COS 地域,如 'ap-guangzhou' 返回: dict: 查询到的数据字典，如果没有查到则返回None """ conn = None cursor = None try: # 连接数据库 conn = pymssql.connect( server=ip, port=port, user=username, password=password, database=database ) cursor = conn.cursor(as_dict=True) # 查询一条符合条件的记录 query_sql = """ SELECT TOP 1 Id, Attachment, VATBusinessRecordId, BusinessCode, Code, LegalPersonFullNamePinYin, LegalPersonIdNumber, LegalPersonIDCardType, LegalPersonCountry, LegalPersonGender, LegalPersonBirthDate, LegalPersonCityEngName, LegalPersonAddressProvinceEn, CompanyAddressLine1En, CompanyAddressLine2En, CityEngName, CompanyAddressPostcode, CompanyAddressProvinceEn, CompanyCountry, CompanyCountryCode, LegalSignedFile, CreatedAt FROM AsyncVATTasks WHERE TaskStatus = 2 AND pdf030_result IN (0, 1) ORDER BY id asc """ cursor.execute(query_sql) result = cursor.fetchone() # 如果查询到数据，则验证字段完整性 if result: # 定义必须非空的字段列表（排除LegalSignedFile） required_fields = [ 'VATBusinessRecordId', 'LegalPersonFullNamePinYin', 'LegalPersonIdNumber', 'LegalPersonIDCardType', 'LegalPersonCountry', 'LegalPersonGender', 'LegalPersonBirthDate', 'LegalPersonCityEngName', 'LegalPersonAddressProvinceEn', 'CompanyAddressLine1En', 'CityEngName', 'CompanyAddressPostcode', 'CompanyAddressProvinceEn', 'CompanyCountry', 'CompanyCountryCode', 'CreatedAt' ] # 检查所有必填字段是否为空 has_empty_field = False for field in required_fields: if result.get(field) is None or result.get(field) == '': has_empty_field = True break # 如果有必填字段为空，设置pdf030_result为3并提交更新 if has_empty_field: update_sql = """ UPDATE AsyncVATTasks SET pdf030_result = 3 WHERE Id = %s """ cursor.execute(update_sql, (result.get('Id'),)) conn.commit() return None # API流程任务业务编码写入 Code 列（BusinessCode 列为空）， # 此处兜底到 BusinessCode，保持输出字段名不变 if not result.get('BusinessCode') and result.get('Code'): result['BusinessCode'] = result['Code'] # 处理结果中的各种字段，传入数据库连接参数与 COS 凭证用于LegalSignedFile处理 process_result_fields(result, ip, port, username, password, secret_id, secret_key, bucket, region) # 更新pdf030_result为1 update_sql = """ UPDATE AsyncVATTasks SET pdf030_result = 1 WHERE id = %s """ cursor.execute(update_sql, (result.get('id'),)) # 提交更新 conn.commit() return result else: return None except Exception as e: print(f"数据库操作错误: {e}") raise finally: # 关闭游标和连接 if cursor: cursor.close() if conn: conn.close()