-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathbasic_example.py
More file actions
276 lines (237 loc) · 10.6 KB
/
Copy pathbasic_example.py
File metadata and controls
276 lines (237 loc) · 10.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
from tax.invoice import InvoiceClient
import time
import traceback # 新增导入
import json
import redis
import qrcode
import io
import base64
from PIL import Image
import time
from inputimeout import inputimeout, TimeoutOccurred
# pip install inputimeout
from datetime import datetime, timedelta
APP_KEY = ''
APP_SECRET = ''
# 配置信息
nsrsbh = "" # 统一社会信用代码
title = "" # 名称（营业执照）
username = "" # 手机号码（电子税务局）
password = "" # 个人用户密码（电子税务局）
type = "6"# 6基础 7标准
xhdwdzdh = "重庆市渝北区龙溪街道丽园路2号XXXX 1325580XXXX" # 地址和电话 空格隔开
xhdwyhzh = "工商银行XXXX 15451211XXXX" #开户行和银行账号 空格隔开
token = ""
debug = True
# 连接Redis 安装pip install redis
r = redis.Redis(
host='127.0.0.1',
port=6379,
password="test123456", # 如果有密码，填写密码
db=0, # 使用第0个数据库
decode_responses=True # 自动将Redis返回的字节转为字符串
)
# 初始化客户端
client = InvoiceClient(
app_key=APP_KEY,
app_secret=APP_SECRET,
base_url="https://api.fa-piao.com",
debug=debug
)
#字符串转二维码图片base64
def str_to_qr_base64(data: str) -> str:
# 创建二维码对象
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=10,
border=4,
)
qr.add_data(data)
qr.make(fit=True)
# 生成图像（PIL Image）
img = qr.make_image(fill_color="black", back_color="white")
# 将图像保存到内存中的字节流
buffer = io.BytesIO()
img.save(buffer, format="PNG")
# 获取 Base64 编码
img_base64 = base64.b64encode(buffer.getvalue()).decode("utf-8")
return img_base64
#字符串转二维码图片 在命令行打印
def print_qr_code(data):
# 创建二维码对象
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L, # 低容错率，生成的图案更稀疏
box_size=1, # 字符画中通常设为1
border=1, # 边框宽度
)
qr.add_data(data)
qr.make(fit=True)
# 使用终端输出模式
# qrcode 库支持将二维码输出为 ANSI 字符
qr.print_ascii(invert=True) # invert=True 会让背景变黑，码变白（适合深色背景终端）
def blue_invoice():
# 开票税额计算demo
# @link https://github.com/fapiaoapi/invoice-sdk-python/blob/master/examples/tax_example.py
invoice_params = {
"fpqqlsh": APP_KEY + str(int(time.time() * 1000)), # 建议用你的订单号
"fplxdm": "82",
"kplx": "0",
"xhdwsbh": nsrsbh,
"username": username,
"xhdwmc": title,
"xhdwdzdh": xhdwdzdh,
"xhdwyhzh": xhdwyhzh,
"ghdwmc": "个人",
# "ghdwsbh": "91330100563015xxxx",
"zsfs": "0",
"bz": "",
# 添加商品明细
"fyxm": [
{
"fphxz": "0",
"spmc": "*研发和技术服务*技术服务费",
"je": 114.68,
"sl": 0.01,
"se": 1.14,
"hsbz": 1,
"spbm": "3040105000000000000"
}
],
# 合计金额
"hjje": 113.54,
"hjse": 1.14,
"jshj": 114.68
}
# 数电蓝票开具接口 文档
# @link https://fa-piao.com/doc.html#api6?source=github
return client.invoice.blue_invoice(**invoice_params)
def get_pdf_ofd_xml(fphm):
pdf_params = {
"downflag": "4",
"nsrsbh": nsrsbh,
"username": username,
"fphm": fphm
}
# 获取销项数电版式文件 文档 PDF/OFD/XML
# @link https://fa-piao.com/doc.html#api7?source=github
pdf_response = client.invoice.get_pdf_ofd_xml(**pdf_params)
if pdf_response.get("code") == 200:
print(pdf_response.get("data"))
def get_token(force_update=False):
key = nsrsbh + "@" + username + "@TOKEN"
if force_update:
# 获取授权Token文档
# @link https://fa-piao.com/doc.html#api1?source=github
auth_response = client.auth.get_authorization(nsrsbh,type)
# auth_response = client.auth.get_authorization(nsrsbh,type,username,password)
if auth_response.get("code") == 200:
print(f"授权成功，Token: {auth_response.get('data', {}).get('token')}")
token = auth_response.get('data', {}).get('token')
r.set(key, token, ex=3600*24*30) # 过期时间30天
client.auth.set_token(token)
else:
token = r.get(key)
if token:
client.auth.set_token(token)
else:
get_token(force_update=True)
try:
# 一 获取授权token 可从缓存redis中获取Token
get_token()
# 前端模拟数电发票/电子发票开具 (蓝字发票)
# @link https://fa-piao.com/fapiao.html?source=github
# 二 开具蓝票
invoice_response = blue_invoice()
match invoice_response.get("code"):
case 200:
print("获取pdf/ofd/xml")
print(f"data: {invoice_response.get('data')}")
# 三 获取pdf/ofd/xml
get_pdf_ofd_xml(invoice_response.get('data', {}).get('Fphm'))
case 420:
print("登录(短信认证)")
# 登录数电发票平台 短信
# 前端模拟短信认证弹窗
# @link https://fa-piao.com/fapiao.html?action=sms&source=github
# 1 发短信验证码
# @link https://fa-piao.com/doc.html#api2?source=github
login_response = client.auth.login_dppt(nsrsbh=nsrsbh, username=username, password=password)
if login_response.get("code") == 200:
deadline = (datetime.now() + timedelta(seconds=300)).strftime("%Y-%m-%d %H:%M:%S")
try:
# 设置 timeout=300 表示 300 秒
sms = inputimeout(prompt=f"请在300秒内({deadline}前)输入验证码: ", timeout=300)
print(f"\n你输入了: {sms}")
# 2 输入验证码
# @link https://fa-piao.com/doc.html#api2?source=github
login_response2 = client.auth.login_dppt(nsrsbh=nsrsbh, username=username, password=password, sms=sms)
if login_response2.get("code") == 200:
print(login_response2.get("data"))
print("\n短信验证成功")
print("\n再次调用blue_invoice")
invoice_response = blue_invoice()
get_pdf_ofd_xml(invoice_response.get('data', {}).get('Fphm') )
except TimeoutOccurred:
print("\n短信认证输入超时!")
case 430:
print("人脸认证")
# 前端模拟人脸认证弹窗
# @link https://fa-piao.com/fapiao.html?action=face&source=github
# 1 获取人脸二维码
# @link https://fa-piao.com/doc.html#api3?source=github
qr_code_response = client.face.get_face_img(nsrsbh=nsrsbh, username=username, type="1")
ewmly = qr_code_response.get("data", {}).get("ewmly")
if ewmly == "swj":
print("\n请使用电子税务局app扫码")
print("\n成功做完人脸认证,请输入数字 1")
# ewm自己生成二维码 安装pip install qrcode[pil]
if qr_code_response.get("data", {}).get("ewm") is not None and len(qr_code_response.get("data", {}).get("ewm")) < 500:
# ewm = qr_code_response.get("data", {}).get("ewm")
# qr_base64 = str_to_qr_base64(ewm)
# qr_code_response.get("data", {}).update({"ewm": qr_base64})
# print(qr_base64)
# 前端使用示例: base64Uri = 'data:image/png;base64,' + qr_base64
print_qr_code(qr_code_response.get("data", {}).get("ewm"))
deadline = (datetime.now() + timedelta(seconds=300)).strftime("%Y-%m-%d %H:%M:%S")
try:
# 设置 timeout=300 表示 300 秒
input_num = inputimeout(prompt=f"请在300秒内({deadline}前)输入: ", timeout=300)
print(f"\n你输入了: {input_num}")
# 2 认证完成后 获取人脸二维码认证状态
# @link https://fa-piao.com/doc.html#api4?source=github
rzid = qr_code_response.get("data", {}).get("rzid")
face_status_response = client.face.get_face_state(nsrsbh=nsrsbh, rzid=rzid, username=username, type="1")
print("code: " + str(face_status_response.get("code")))
print("data: " + str(face_status_response.get("data")))
if face_status_response.get("data") is not None:
slzt = face_status_response.get("data", {}).get("slzt")
if slzt == "1":
print("\n人脸未认证")
elif slzt == "2":
print("\n人脸认证成功")
print("\n请再次调用blue_invoice")
invoice_response = blue_invoice()
get_pdf_ofd_xml(invoice_response.get('data', {}).get('Fphm'))
elif slzt == "3":
print("\n人脸认证二维码过期-->重新获取人脸二维码")
except TimeoutOccurred:
print("\n人脸认证输入超时!")
elif ewmly == "grsds":
print("个人所得税app扫码") # ewm是二维码的base64
case 401:
# 重新授权
print(f"{invoice_response.get('code')}授权失败:{invoice_response.get('msg')}")
print("重新授权 获取token 缓存到redis")
get_token(True)
invoice_response = blue_invoice()
get_pdf_ofd_xml(invoice_response.get('data', {}).get('Fphm'))
case _:
print(f"{invoice_response.get('code')}异常: {invoice_response.get('msg')}")
except Exception as e:
# 打印完整堆栈信息（包含行号）
traceback.print_exc() # 新增堆栈跟踪
# 添加上下文信息（可选）
print(f"\n错误发生时参数状态：")
print(f"当前时间戳: {int(time.time())}")