liubiren
/
Python
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

1

This commit is contained in:
liubiren 2026-03-18 15:03:35 +08:00
parent 4f0739fa92
commit 0f8c56e72a
3 changed files with 16 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -1,4 +1,3 @@
__pycache__/ __pycache__/
票据理赔自动化/.DS_Store
.DS_Store .DS_Store
.gitignore .gitignore

5
utils/authenticator.py
View File

@ -54,7 +54,7 @@ class Authenticator:
f"初始化所有服务商的访问凭证发生异常：{str(exception)}" f"初始化所有服务商的访问凭证发生异常：{str(exception)}"
) from exception ) from exception
# 初始化请求客户端 # 实例化请求客户端
self.request = Request() self.request = Request()
def get_token(self, servicer: str) -> Optional[str]: def get_token(self, servicer: str) -> Optional[str]:
@ -101,7 +101,6 @@ class Authenticator:
fp=file, fp=file,
ensure_ascii=False, ensure_ascii=False,
) )
return token return token
def _refresh_inspirvision_certification(self) -> Tuple[str, float]: def _refresh_inspirvision_certification(self) -> Tuple[str, float]:
@ -167,4 +166,4 @@ class Authenticator:
return ( return (
response["tenant_access_token"], response["tenant_access_token"],
time.time() + response["expire"], time.time() + response["expire"],
) ) # https://open.feishu.cn/document/server-docs/api-call-guide/calling-process/get-access-token 飞书获取访问凭证接口

32
utils/feishu.py
View File

@ -8,8 +8,8 @@ from email.policy import default
from email.utils import parsedate_to_datetime from email.utils import parsedate_to_datetime
from imaplib import IMAP4_SSL from imaplib import IMAP4_SSL
import re import re
import time from time import sleep, time
from typing import Dict, Any from typing import Any, Dict
import pandas import pandas
@ -21,9 +21,9 @@ class Feishu:
"""飞书客户端""" """飞书客户端"""
def __init__(self): def __init__(self):
# 初始化认证器 # 实例化认证器
self.authenticator = Authenticator() self.authenticator = Authenticator()
# 初始化请求客户端 # 实例化请求客户端
self.http_client = Request() self.http_client = Request()
def _get_headers(self) -> Dict[str, Any]: def _get_headers(self) -> Dict[str, Any]:
@ -36,23 +36,19 @@ class Feishu:
@staticmethod @staticmethod
def get_verification_code(): def get_verification_code():
try: try:
# 执行时间戳 # 当前时间戳
execute_timestamp = time.time() current_timestamp = time()
# 超时时间戳
timeout_timestamp = execute_timestamp + 65
# 建立加密IMAP连接 # 建立加密IMAP连接
server = IMAP4_SSL("imap.feishu.cn", 993) connection = IMAP4_SSL(host="imap.feishu.cn", port=993)
# 登录 # 登录
server.login("mars@liubiren.cloud", "a2SfPUgbKDmrjPV2") connection.login(user="mars@liubiren.cloud", password="aJBZSZzhQN13M11K")
while True: while True:
# 若当前时间戳大于超时时间戳则返回NONE if time() <= current_timestamp + 120:
if time.time() <= timeout_timestamp: sleep(5)
# 等待10秒 # 选择邮箱文件夹（邮箱验证码）
time.sleep(10) connection.select(mailbox="&kK57sZqMi8F4AQ-")
# 选择文件夹（邮箱验证码）
server.select("&kK57sZqMi8F4AQ-")
# noinspection PyBroadException
try: try:
# 获取最后一封邮件索引，server.search()返回数据类型为元组，第一个元素为查询状态，第二个元素为查询结果（邮件索引字节串的列表）；然后，从列表获取字节串并分割取最后一个，作为最后一封邮件索引 # 获取最后一封邮件索引，server.search()返回数据类型为元组，第一个元素为查询状态，第二个元素为查询结果（邮件索引字节串的列表）；然后，从列表获取字节串并分割取最后一个，作为最后一封邮件索引
index = server.search(None, "ALL")[1][0].split()[-1] index = server.search(None, "ALL")[1][0].split()[-1]
@ -132,7 +128,7 @@ class Feishu:
response = self.http_client.post( response = self.http_client.post(
url=url_next, url=url_next,
headers=self._headers(), headers=self._get_headers(),
json={"field_names": field_names, "filter": filter_conditions}, json={"field_names": field_names, "filter": filter_conditions},
) )