246 lines
8.4 KiB
Python
246 lines
8.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
飞书客户端模块
|
||
"""
|
||
|
||
import re
|
||
import time
|
||
from email.parser import BytesParser
|
||
from email.policy import default
|
||
from email.utils import parsedate_to_datetime
|
||
from imaplib import IMAP4_SSL
|
||
|
||
import pandas
|
||
from request import Request
|
||
from authenticator import Authenticator
|
||
|
||
|
||
class Feishu:
|
||
"""飞书客户端"""
|
||
|
||
def __init__(self):
|
||
self.authenticator = Authenticator()
|
||
self.http_client = Request()
|
||
|
||
def _headers(self):
|
||
"""请求头"""
|
||
# 装配飞书访问凭证
|
||
return {
|
||
"Authorization": f"Bearer {self.authenticator.get_token(servicer='feishu')}",
|
||
}
|
||
|
||
@staticmethod
|
||
def get_verification_code():
|
||
try:
|
||
# 执行时间戳
|
||
execute_timestamp = time.time()
|
||
# 超时时间戳
|
||
timeout_timestamp = execute_timestamp + 65
|
||
# 建立加密IMAP连接
|
||
server = IMAP4_SSL("imap.feishu.cn", 993)
|
||
# 登录
|
||
server.login("mars@liubiren.cloud", "a2SfPUgbKDmrjPV2")
|
||
|
||
while True:
|
||
# 若当前时间戳大于超时时间戳则返回NONE
|
||
if time.time() <= timeout_timestamp:
|
||
# 等待10秒
|
||
time.sleep(10)
|
||
# 选择文件夹（邮箱验证码）
|
||
server.select("&kK57sZqMi8F4AQ-")
|
||
# noinspection PyBroadException
|
||
try:
|
||
# 获取最后一封邮件索引，server.search()返回数据类型为元组，第一个元素为查询状态，第二个元素为查询结果（邮件索引字节串的列表）；然后，从列表获取字节串并分割取最后一个，作为最后一封邮件索引
|
||
index = server.search(None, "ALL")[1][0].split()[-1]
|
||
# 获取最后一封邮件内容并解析，server.fetch()返回数据类型为元组，第一个元素为查询状态，第二个元素为查询结果（邮件内容字节串的列表）；然后，从列表获取字节串并解析正文
|
||
# noinspection PyUnresolvedReferences
|
||
contents = BytesParser(policy=default).parsebytes(
|
||
server.fetch(index, "(RFC822)")[1][0][1]
|
||
)
|
||
# 遍历邮件内容，若正文内容类型为纯文本或HTML则解析发送时间和验证码
|
||
for content in contents.walk():
|
||
if (
|
||
content.get_content_type() == "text/plain"
|
||
or content.get_content_type() == "text/html"
|
||
):
|
||
# 邮件发送时间戳
|
||
# noinspection PyUnresolvedReferences
|
||
send_timestamp = parsedate_to_datetime(
|
||
content["Date"]
|
||
).timestamp()
|
||
# 若邮件发送时间戳大于执行时间戳则解析验证码并返回
|
||
if (
|
||
execute_timestamp
|
||
> send_timestamp
|
||
>= execute_timestamp - 35
|
||
):
|
||
# 登出
|
||
server.logout()
|
||
# 解析验证码
|
||
return re.search(
|
||
r"【普康健康】您的验证码是：(\d+)",
|
||
content.get_payload(decode=True).decode(),
|
||
).group(1)
|
||
|
||
# 若文件夹无邮件则继续
|
||
except:
|
||
pass
|
||
|
||
# 若超时则登出
|
||
else:
|
||
server.logout()
|
||
return None
|
||
|
||
except Exception:
|
||
raise RuntimeError("获取邮箱验证码发生其它异常")
|
||
|
||
# 查询多维表格记录，单次最多查询500条记录
|
||
@restrict(refill_rate=5, max_tokens=5)
|
||
def query_bitable_records(
|
||
self,
|
||
bitable: str,
|
||
table_id: str,
|
||
field_names: Optional[list[str]] = None,
|
||
filter_conditions: Optional[dict] = None,
|
||
) -> pandas.DataFrame:
|
||
# 先查询多维表格记录，在根据字段解析记录
|
||
# 装配多维表格查询记录地址
|
||
url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{bitable}/tables/{table_id}/records/search?page_size=20"
|
||
|
||
response = self.http_client.post(
|
||
url=url,
|
||
headers=self._headers(),
|
||
json={"field_names": field_names, "filter": filter_conditions},
|
||
)
|
||
|
||
# 响应业务码为0则定义为响应成功
|
||
assert response.get("code") == 0, "查询多维表格记录发生异常"
|
||
|
||
# 多维表格记录
|
||
records = response.get("data").get("items")
|
||
|
||
# 检查响应中是否包含还有下一页标识，若有则继续请求下一页
|
||
while response.get("data").get("has_more"):
|
||
|
||
url_next = url + "&page_token={}".format(
|
||
response.get("data").get("page_token")
|
||
)
|
||
|
||
response = self.http_client.post(
|
||
url=url_next,
|
||
headers=self._headers(),
|
||
json={"field_names": field_names, "filter": filter_conditions},
|
||
)
|
||
|
||
assert response.get("code") == 0, "查询多维表格记录发生异常"
|
||
|
||
# 合并记录
|
||
records.append(response.get("data").get("items"))
|
||
|
||
# 装配多维表格列出字段地址
|
||
url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{bitable}/tables/{table_id}/fields?page_size=20"
|
||
|
||
response = self.http_client.get(
|
||
url=url,
|
||
headers=self._headers(),
|
||
)
|
||
|
||
assert response.get("code") == 0, "列出多维表格字段发生异常"
|
||
|
||
# 多维表格字段
|
||
fields = response.get("data").get("items")
|
||
|
||
while response.get("data").get("has_more"):
|
||
|
||
url_next = url + "&page_token={}".format(
|
||
response.get("data").get("page_token")
|
||
)
|
||
|
||
response = self.http_client.get(
|
||
url=url_next,
|
||
headers=self._headers(),
|
||
)
|
||
|
||
assert response.get("code") == 0, "列出多维表格字段发生异常"
|
||
|
||
fields.append(response.get("data").get("items"))
|
||
|
||
# 字段映射
|
||
field_mappings = {}
|
||
|
||
for field in fields:
|
||
|
||
# 字段名
|
||
field_name = field["field_name"]
|
||
|
||
# 根据字段类型匹配
|
||
match field["type"]:
|
||
|
||
case 1005:
|
||
|
||
field_type = "主键"
|
||
|
||
case 1:
|
||
|
||
field_type = "文本"
|
||
|
||
case 3:
|
||
|
||
field_type = "单选"
|
||
|
||
case 2:
|
||
|
||
# 数字、公式字段的显示格式
|
||
match field["property"]["formatter"]:
|
||
|
||
case "0":
|
||
|
||
field_type = "整数"
|
||
|
||
case _:
|
||
|
||
raise ValueError("未设置数字、公式字段的显示格式")
|
||
|
||
case _:
|
||
|
||
raise ValueError("未设置字段类型")
|
||
|
||
# noinspection PyUnboundLocalVariable
|
||
field_mappings.update({field_name: field_type})
|
||
|
||
# 记录数据体
|
||
records_data = []
|
||
|
||
# 解析记录
|
||
for record in records:
|
||
|
||
# 单条记录数据体
|
||
record_data = {}
|
||
|
||
for field_name, content in record["fields"].items():
|
||
|
||
match field_mappings[field_name]:
|
||
|
||
case "主键" | "单选" | "整数":
|
||
|
||
record_data.update({field_name: content})
|
||
|
||
case "文本":
|
||
|
||
# 若存在多行文本则拼接
|
||
fragments_content = ""
|
||
|
||
for fragment_content in content:
|
||
|
||
fragments_content += fragment_content["text"]
|
||
|
||
record_data.update({field_name: fragments_content})
|
||
|
||
case _:
|
||
|
||
raise ValueError("未设置字段解析方法")
|
||
|
||
records_data.append(record_data)
|
||
|
||
return pandas.DataFrame(records_data)
|