# 基礎(chǔ)設(shè)施層
import pandas as pd        # 數(shù)據(jù)處理
import requests           # 網(wǎng)絡(luò)請(qǐng)求
import logging            # 錯(cuò)誤追蹤
from datetime import datetime # 時(shí)間處理
 
# 配置層
logging配置 → API密鑰配置
 
# 功能層
get_access_token() → 獲取系統(tǒng)通行證
insert_to_feishu() → 內(nèi)容插入邏輯
 
# 執(zhí)行層
main() → 流程控制器

response = requests.post(url, json=data)  # 發(fā)送驗(yàn)證請(qǐng)求
return response.json()["tenant_access_token"]  # 提取通行證

blocks = requests.get(blocks_url).json()["data"]["items"]

last_block_id = blocks[-1]["block_id"]  # 列表末尾元素

# 添加醒目時(shí)間標(biāo)簽
時(shí)間戳 = datetime.now().strftime(...)
# 將Excel轉(zhuǎn)為Markdown表格
markdown_table = df.to_markdown(index=False)

requests.post(insert_url, json={
    "content": "**時(shí)間標(biāo)簽** + 表格數(shù)據(jù)",
    "position": "after"  # 放在最后
})

logging.basicConfig(
    filename='feishu_sync.log',  # 記錄文件
    level=logging.INFO,          # 記錄級(jí)別
    format='時(shí)間-級(jí)別-信息'       # 記錄格式
)

try:
    發(fā)送請(qǐng)求...
except Exception as e:
    logging.error(f"插入失敗: {e}")  # 記錄錯(cuò)誤快照

def get_access_token():
    """獲取飛書API的access_token"""
    # 定義API請(qǐng)求地址
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    # 設(shè)置HTTP請(qǐng)求頭（告訴服務(wù)器我們發(fā)送的是JSON數(shù)據(jù)）
    headers = {"Content-Type": "application/json"}
"""
服務(wù)器會(huì)根據(jù) Content-Type 的值，調(diào)用對(duì)應(yīng)的解析方式。例如：
如果是 application/json，服務(wù)器會(huì)用 JSON 解析器處理數(shù)據(jù)。
如果是 application/x-www-form-urlencoded，服務(wù)器會(huì)按表單格式解析數(shù)據(jù)。
"""
    # 準(zhǔn)備請(qǐng)求數(shù)據(jù)（飛書應(yīng)用的ID和密鑰）
    data = {"app_id": APP_ID, "app_secret": APP_SECRET}
    
    try:
        # 發(fā)送POST請(qǐng)求到飛書服務(wù)器
        response = requests.post(url, headers=headers, json=data)
        # 檢查響應(yīng)狀態(tài)碼（如果不是200，會(huì)拋出異常）
        response.raise_for_status()
        # 解析返回的JSON數(shù)據(jù)，提取訪問(wèn)令牌
        return response.json()["tenant_access_token"]
    
    except Exception as e:
        # 記錄錯(cuò)誤日志
        logging.error(f"獲取access_token失敗: {e}")
        # 失敗時(shí)返回None
        return None

    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()["tenant_access_token"]
 
    except Exception as e:
        logging.error(f"獲取access_token失敗: {e}")
        return None

access_token = get_access_token()

blocks_url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{DOCUMENT_ID}/blocks"
headers = {"Authorization": f"Bearer {access_token}"}
blocks_response = requests.get(blocks_url, headers=headers)

 headers = {"Authorization": f"Bearer {access_token}"}

# 假設(shè) access_token = "abc123xyz"
headers = {"Authorization": "Bearer abc123xyz"}

# 發(fā)送請(qǐng)求時(shí)附加 headers
response = requests.get(url, headers=headers)

blocks_response = requests.get(blocks_url, headers=headers)

blocks = blocks_response.json()["data"]["items"]

{
  "code": 0,
  "data": {
    "items": [
      {"block_id": "xxxx", "content": "文本內(nèi)容1"},
      {"block_id": "yyyy", "content": "文本內(nèi)容2"}
    ]
  }
}

data = {
    "code": 0,
    "message": "成功",
    "data": {
        "items": [
            {"姓名": "張三", "年齡": 25, "城市": "北京"},
            {"姓名": "李四", "年齡": 30, "城市": "上海"}
        ]
    }
}
 
# 提取表格中的每一行（即 items 列表）
items = data["data"]["items"]

blocks = [
    {"block_id": "段落1", "text": "這是第一段內(nèi)容"},
    {"block_id": "圖片1", "url": "https://example.com/image.jpg"}
]

last_block_id = blocks[-1]["block_id"] if blocks else None

insert_url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{DOCUMENT_ID}/blocks/{last_block_id}/insert"
data = {
    "content": f"**數(shù)據(jù)更新時(shí)間：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}**\n{content}",
    "position": "after"
}
insert_response = requests.post(insert_url, headers=headers, json=data)

 insert_response = requests.post(insert_url, headers=headers, json=data)
        insert_response.raise_for_status()
        logging.info("數(shù)據(jù)插入成功！")

try:
    # ...請(qǐng)求代碼...
except Exception as e:
    logging.error(f"插入數(shù)據(jù)失敗: {e}")
    return False

def main():
    """主函數(shù)：讀取Excel并同步到飛書"""
    try:
        # 讀取Excel數(shù)據(jù)
        df = pd.read_excel("data.xlsx")  # 替換為你的Excel路徑
        markdown_table = df.to_markdown(index=False)
 
        # 插入到飛書文檔
        if insert_to_feishu(markdown_table):
            print("同步成功！")
        else:
            print("同步失敗，請(qǐng)查看日志文件。")
    except Exception as e:
        logging.error(f"主流程異常: {e}")
markdown_table = df.to_markdown(index=False) 

import pandas as pd
 
data = {
        "Name": ["Alice", "Bob",  "Charlie"],
    "Age":  [25,  30, 35],
        "City": ["New York",  "London",  "Tokyo"]
}
df  =  pd.DataFrame(data)

markdown_table  =  df.to_markdown(index=False)
print(markdown_table)

#  示例：讀取  CSV  并處理數(shù)據(jù)
data = pd.read_csv("data.csv")
filtered_data  =  data[data["age"] >  18]
filtered_data.to_csv("adults.csv")

#  示例：調(diào)用 API  獲取數(shù)據(jù)
response  =  requests.get("https://api.example.com/data")
if  response.status_code  ==  200:
        data  = response.json()
     print(data["results"])

#  示例：記錄程序運(yùn)行狀態(tài)
logging.basicConfig(level=logging.INFO)
logging.info("程序啟動(dòng)")
try:
        #  某些操作
except Exception as  e:
        logging.error(f"發(fā)生錯(cuò)誤:  {e}")

#  示例：生成帶時(shí)間戳的日志
current_time =  datetime.now().strftime("%Y-%m-%d  %H:%M:%S")
print(f"操作時(shí)間:  {current_time}")

import pandas as pd
import requests
import logging
from datetime import datetime
 
# 配置日志記錄
logging.basicConfig(
    filename='feishu_sync.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
"""
1.logging.basicConfig(...)
logging 是 Python 標(biāo)準(zhǔn)庫(kù)中的日志模塊，用于記錄程序的運(yùn)行狀態(tài)，方便調(diào)試和排查問(wèn)題。
basicConfig() 方法用于配置日志系統(tǒng)的基本設(shè)置。
2.filename='feishu_sync.log'
指定日志輸出的文件名為 feishu_sync.log，所有日志信息都會(huì)被寫入該文件。
這樣可以在程序運(yùn)行后，查看 feishu_sync.log 文件來(lái)了解程序的執(zhí)行情況。
3.level=logging.INFO
設(shè)置日志的最低級(jí)別為 INFO，即只記錄 INFO 及更高級(jí)別的日志（包括 WARNING、ERROR 和 CRITICAL）。
低于 INFO 級(jí)別的 DEBUG 級(jí)日志不會(huì)被記錄。
4.format='%(asctime)s - %(levelname)s - %(message)s'
指定日志的輸出格式，其中：
%(asctime)s：時(shí)間戳，表示日志記錄的時(shí)間（格式如 2025-02-18 12:34:56）。
%(levelname)s：日志級(jí)別，如 INFO、ERROR 等。
%(message)s：具體的日志消息內(nèi)容。
"""
 
 
# 飛書API配置
APP_ID = "your_app_id"
APP_SECRET = "your_app_secret"
DOCUMENT_ID = "Gu4LwF32Ci0OUakZo50cQ8p2njk"  # 目標(biāo)云文檔的ID
 
 
def get_access_token():
    """獲取飛書API的access_token"""
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    headers = {"Content-Type": "application/json"}
    data = {"app_id": APP_ID, "app_secret": APP_SECRET}
    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()["tenant_access_token"]
 
    except Exception as e:
        logging.error(f"獲取access_token失敗: {e}")
        return None
"""
requests.post(url, headers=headers, json=data)：使用 requests 庫(kù)發(fā)送一個(gè) POST 請(qǐng)求到指定的 URL，并攜帶請(qǐng)求頭和請(qǐng)求體數(shù)據(jù)。json=data 會(huì)自動(dòng)將字典 data 轉(zhuǎn)換為 JSON 格式的字符串。
response.raise_for_status()：檢查響應(yīng)的狀態(tài)碼，如果狀態(tài)碼不是 200（表示請(qǐng)求成功），則拋出 HTTPError 異常。
response.json()["tenant_access_token"]：將響應(yīng)的內(nèi)容解析為 JSON 格式，并返回其中的 tenant_access_token 字段，即獲取到的租戶訪問(wèn)令牌。
except Exception as e:：捕獲所有異常，并將異常對(duì)象賦值給變量 e。
logging.error(f"獲取access_token失敗: {e}")：使用 logging 模塊記錄錯(cuò)誤信息，方便后續(xù)排查問(wèn)題。
return None：如果發(fā)生異常，返回 None 表示獲取令牌失敗。
"""
 
def insert_to_feishu(content):
    """向飛書文檔插入內(nèi)容"""
    access_token = get_access_token()
    if not access_token:
        return False
 
    # 步驟1：獲取文檔塊列表，找到最后一個(gè)塊的位置
    blocks_url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{DOCUMENT_ID}/blocks"
    headers = {"Authorization": f"Bearer {access_token}"}
 
    try:
        # 獲取文檔結(jié)構(gòu)
        blocks_response = requests.get(blocks_url, headers=headers)
        blocks_response.raise_for_status()
        blocks = blocks_response.json()["data"]["items"]
 
        # 找到最后一個(gè)塊的ID（用于插入位置）
        last_block_id = blocks[-1]["block_id"] if blocks else None
 
        # 步驟2：插入新內(nèi)容到文檔末尾
        insert_url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{DOCUMENT_ID}/blocks/{last_block_id}/insert"
        data = {
            "content": f"**數(shù)據(jù)更新時(shí)間：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}**\n{content}",
            "position": "after"  # 在最后一個(gè)塊后插入
        }
        insert_response = requests.post(insert_url, headers=headers, json=data)
        insert_response.raise_for_status()
        logging.info("數(shù)據(jù)插入成功！")
        return True
    except Exception as e:
        logging.error(f"插入數(shù)據(jù)失敗: {e}")
        return False
 
 
def main():
    """主函數(shù)：讀取Excel并同步到飛書"""
    try:
        # 讀取Excel數(shù)據(jù)
        df = pd.read_excel("data.xlsx")  # 替換為你的Excel路徑
        markdown_table = df.to_markdown(index=False)
 
        # 插入到飛書文檔
        if insert_to_feishu(markdown_table):
            print("同步成功！")
        else:
            print("同步失敗，請(qǐng)查看日志文件。")
    except Exception as e:
        logging.error(f"主流程異常: {e}")
 
 
if __name__ == "__main__":
    main()