python调用国税乐企直连接口开数电票之查询税收分类编码信息

背景

通过国税官方开放乐企平台接口, 实现了在EPR系统内直接开票. 无需通过任何第三方系统.逐步更新到CSDN专栏分享大家参考.

接口说明

• 定时获取可用税收分类编码，绑定ERP内部的编码, 使开票时能根据商品匹配到对应的税收分类编码…针对不同编码维护了 18 类增值税特殊管理内容（如简易征收、免税等）及对应优惠政策。开具数电票时，需按需求选择优惠政策代码和税率，不同优惠政策对应不同的优惠政策标识填写规则与税率限制.

直接上代码

#!/usr/bin/env python
# encoding: utf-8
# @author: Evan@chengwenit.com
# 乐企数电票税收分类编码信息拉取到ERPimport cjc_cfg as cfg
from lq_request_util import LQRequestUtil
from cjc_oracle_client import OracleDBClient# 定义数据库连接
db = OracleDBClient()
db.connect()# 构建工具类实例
requester = LQRequestUtil(base_url=cfg.url_front,headers=cfg.hjheaders,key=cfg.hjlqkey
)# 从ERP数据库获取全部的税收分类编码和最近更新的时间戳
selectsql0 = """ select SPHFWSSFLHBBM,SJC from CJC_LQ_SSFLBM_DTL order by SJC DESC """
print(selectsql0)
ssflxx = db.execute_query(selectsql0)
ssflbm_list = [d.get("SPHFWSSFLHBBM") for d in ssflxx if "SPHFWSSFLHBBM" in d]
sjc = ssflxx[0]["SJC"] if ssflxx else '19700101000000'# 定义请求数据
taxno = cfg.hjxfxx["taxno"]
request_data = {"nsrsbh":taxno,"sjc":sjc}# 调用接口
try:result = requester.post(cfg.url_after["flbm"], request_data, verify_ssl=False)if result.get("Data") is not None:print(str(result)[:1000])if result["Data"]["returncode"] == '00':for item in result["Data"]["ssbmList"]:# 判断税收分类编码是否存在,存在则更新,不存在则插入if item["sphfwssflhbbm"] in ssflbm_list:updatesql0 = f""" update CJC_LQ_SSFLBM_DTL set FJBM='{item.get("fjbm",'')}',TDYSLXDM='{item.get("tdyslxdm",'').lstrip('0')}',CEZS='{item.get("cezs",'')}',BZS='{item.get("bzs",'')}',JZJTDM='{item.get("jzjtdm",'')}',SPHFWMC='{item.get("sphfwmc",'')}',SPHFWFLJC='{item.get("sphfwfljc",'')}',SM='{item.get("sm",'')[:1000]}',ZZSSL='{item.get("zzssl",'')}',ZSL='{item.get("zsl",'')}',ZZSTSGL='{item.get("zzstsgl",'')}',ZZSZCYJ='{item.get("zzszcyj",'')}',ZZSTSNRDM='{item.get("zzstsnrdm",'')}',XFSGL='{item.get("xfsgl",'')}',XFSZCYJ='{item.get("xfszcyj",'')}',XFSTSNRDM='{item.get("xfstsnrdm",'')}',SFHZX='{item.get("sfhzx",'')}',QYRQ='{item.get("qyrq",'')}',TYRQ='{item.get("tyrq",'')}',SJC='{result["Data"]["sjc"]}'where SPHFWSSFLHBBM='{item["sphfwssflhbbm"]}' """db.execute_non_query(updatesql0)else:insertsql0 = f""" insert into CJC_LQ_SSFLBM_DTL(SPHFWSSFLHBBM, FJBM, TDYSLXDM, CEZS, BZS, JZJTDM, SPHFWMC, SPHFWFLJC, SM, ZZSSL, ZSL, ZZSTSGL, ZZSZCYJ, ZZSTSNRDM, XFSGL, XFSZCYJ, XFSTSNRDM, SFHZX, QYRQ, TYRQ, SJC) values('{item.get("sphfwssflhbbm",'')}','{item.get("fjbm",'')}','{item.get("tdyslxdm",'').lstrip('0')}','{item.get("cezs",'')}','{item.get("bzs",'')}','{item.get("jzjtdm",'')}','{item.get("sphfwmc",'')}','{item.get("sphfwfljc",'')}','{item.get("sm",'')[:1000]}','{item.get("zzssl",'')}','{item.get("zsl",'')}','{item.get("zzstsgl",'')}','{item.get("zzszcyj",'')}','{item.get("zzstsnrdm",'')}','{item.get("xfsgl",'')}','{item.get("xfszcyj",'')}','{item.get("xfstsnrdm",'')}','{item.get("sfhzx",'')}','{item.get("qyrq",'')}','{item.get("tyrq",'')}','{result["Data"]["sjc"]}') """db.execute_non_query(insertsql0)else:print(result)
except Exception as e:print("请求失败：", e)
finally:#关闭数据库连接db.disconnect()

接口返回示例

成功返回

{
"Response":{
"RequestId": "16位随机数字和字母组合",
"Data":{
"returncode": "00",
"returnmsg": "成功",
"sjc": "时间戳",
"count": "总数量",
"ssbmList": [
{
"fjbm": "父级编码",
"sphfwssflhbbm": "商品和服务税收分类合并编码",
"tdyslxdm": "特定要素标签归类",
"cezs": "差额征税",
"bzs": "不征税",
"jzjtdm": "即征即退代码集合",
"sphfwmc": "商品和服务名称",
"sphfwfljc": "商品和服务分类简称",
"sm": "说明",
"zzssl": "增值税税率",
"zsl": "征收率",
"zzstsgl": "增值税特殊管理",
"zzszcyj": "增值税政策依据",
"zzstsnrdm": "增值税特殊内容代码（增值税减免税政策代码表）",
"xfsgl": "消费税管理",
"xfszcyj": "消费税政策依据",
"xfstsnrdm": "消费税特殊内容代码",
"gjz": "关键字",
"sfhzx": "是否汇总项",
"dytjjbm": "对应统计局编码（2011年版）或国民行业代码",
"hgjcksppm": "海关进出口商品品目",
"qyrq": "启用日期",
"tyrq": "停用日期",
"zspmList": [
{
"hydm": "行业代码",
"dydhy": "对应的行业",
"hydydzspm1": "行业对应的征收品目一",
"hydydzsmc1": "行业对应的征收名称一",
"hydydzspm2": "行业对应的征收品目二",
"hydydzsmc2": "行业对应的征收名称二"
}
]
}
]
}
}
}

失败返回

{
"Response":{
"RequestId": "16位随机数字和字母组合",
"Error":{
"Code": "01”,
"Message": "错误说明"
}
}
}