# 使用的基礎(chǔ)鏡像
 FROM fluent/fluentd:edge-debian
 # 切換用戶為root方便接下來執(zhí)行安裝命令
 USER root
 # 安裝系統(tǒng)依賴
 RUN apt-get update && \
 apt-get install -y curl && \
 rm -rf /var/lib/apt/lists/*

 # 卸載可能存在的高版本 Elasticsearch gem
 RUN gem uninstall elasticsearch elasticsearch-api -a -x || true

 # 安裝指定版本 gem 和 Fluentd Elasticsearch 插件
 RUN gem install elasticsearch -v 8.17.1 --no-document && \
 gem install elasticsearch-api -v 8.17.1 --no-document && \
 gem install fluent-plugin-elasticsearch -v 5.4.3 --no-document
 USER fluent

# 定義本文件中所有要啟動的服務(wù)（容器）
 services:
 # Fluent Bit：輕量收集容器日志
 # 【服務(wù)名】是的，'fluent-bit' 就是這個服務(wù)的名稱。它在 Docker Compose 網(wǎng)絡(luò)內(nèi)部被識別為此名。
 fluent-bit:
 # 使用的鏡像：從 Docker Hub 拉取最新的 Fluent Bit 官方鏡像
 image: fluent/fluent-bit:latest
 # 指定容器啟動后的名稱，通過 `docker ps` 等命令可以看到這個名字
 container_name: fluent-bit
 # 依賴關(guān)系：指定此服務(wù)的啟動依賴于另一個服務(wù) 'fluentd'
 depends_on:
 fluentd:
 # 條件：只有當(dāng) 'fluentd' 服務(wù)通過健康檢查（healthy）后，才會啟動 fluent-bit
 condition: service_healthy
 # 端口映射：將宿主機(jī)的端口映射到容器內(nèi)的端口
 # 格式 - "宿主機(jī)端口:容器端口"
 ports:
 - "2020:2020" # 將容器內(nèi)的 2020 端口（Fluent Bit 的 HTTP Server，常用于健康檢查或監(jiān)控）映射到宿主機(jī)的 2020 端口
 # 數(shù)據(jù)卷掛載：將宿主機(jī)的目錄或文件掛載到容器內(nèi)，實現(xiàn)數(shù)據(jù)持久化或配置注入
 volumes:
 # 將宿主機(jī)的 '/Users/cuihao/docker/EFK/web-logs' 目錄掛載到容器內(nèi)的 '/var/log/' 目錄。
 # Fluent Bit 會監(jiān)控這個目錄下的日志文件變化，并收集新產(chǎn)生的日志。
 #/var/log/metric_log
 #/Users/cuihao/docker/EFK/web-logs
 - /Users/cuihao/docker/EFK/web-logs:/var/log/
 # 將宿主機(jī)的 Fluent Bit 主配置文件掛載到容器內(nèi)，替代鏡像內(nèi)的默認(rèn)配置。
 # 這個文件定義了數(shù)據(jù)輸入（Input）、處理（Parser, Filter）和輸出（Output）的規(guī)則。
 - /Users/cuihao/docker/EFK/fluent-bit/etc/:/fluent-bit/etc/

 # Fluentd：格式化
 # 【服務(wù)名】'fluentd' 是這個服務(wù)的名稱。
 fluentd:
 container_name: fluent
 # 不是使用現(xiàn)成的鏡像，而是通過指定構(gòu)建上下文路徑（E:\EFK\fluentd）來構(gòu)建自定義鏡像。
 # 該路徑下應(yīng)該有一個名為 'Dockerfile' 的文件，默認(rèn)讀取 'Dockerfile' 文件。
 build: /Users/cuihao/docker/EFK/fluentd
 volumes:
 # 將宿主機(jī)上的 Fluentd 配置目錄掛載到容器內(nèi)，使配置變更無需重新構(gòu)建鏡像。
 - /Users/cuihao/docker/EFK/fluentd/conf:/fluentd/etc
 depends_on:
 elasticsearch:
 condition: service_healthy
 ports:
 - "24224:24224" # Fluentd 默認(rèn)的 TCP 端口，用于接收來自 Fluent Bit 或其他客戶端轉(zhuǎn)發(fā)來的日志，這個端口會配置到fluent-bit的OUTPUT中
 - "24224:24224/udp" # Fluentd 默認(rèn)的 UDP 端口，用途同上
 - "24220:24220" # Fluentd 的健康檢查 API 端口
 # 健康檢查配置：Docker 會根據(jù)此規(guī)則判斷容器是否正常啟動
 healthcheck:
 test: ["CMD-SHELL", "curl -fs http://localhost:24220/api/plugins.json || exit 1"] # 檢查健康檢查端點是否返回成功
 interval: 5s # 每 5 秒檢查一次
 timeout: 3s # 每次檢查超時時間為 3 秒
 retries: 5 # 連續(xù)失敗 5 次才標(biāo)記為不健康
 start_period: 10s # 容器啟動后，等待 10 秒再進(jìn)行第一次健康檢查

 # Elasticsearch：存儲日志
 # 【服務(wù)名】'elasticsearch' 是這個服務(wù)的名稱。
 elasticsearch:
 image: docker.elastic.co/elasticsearch/elasticsearch:8.17.1 # 使用 Elastic 官方的 8.17.1 版本鏡像
 container_name: elasticsearch
 hostname: elasticsearch # 設(shè)置容器內(nèi)部的主機(jī)名，在集群中很有用
 environment:
 - discovery.type=single-node # 設(shè)置為單節(jié)點模式，適合開發(fā)和測試
 - xpack.security.enabled=false # 關(guān)閉 X-Pack 安全功能（用戶認(rèn)證、HTTPS等）。生產(chǎn)環(huán)境必須開啟，但測試時關(guān)閉更簡單。
 healthcheck:
 test: ["CMD", "curl", "-f", "http://localhost:9200/_cluster/health"] # 檢查 ES 集群健康狀態(tài) API
 interval: 10s
 retries: 5
 timeout: 5s
 ports:
 - "9200:9200" # 將 ES 的 HTTP REST API 端口映射到宿主機(jī)，方便通過瀏覽器或命令訪問

 # Kibana：日志可視化
 # 【服務(wù)名】'kibana' 是這個服務(wù)的名稱。
 kibana:
 container_name: kibana
 image: docker.elastic.co/kibana/kibana:8.17.1 # Kibana 版本需要與 Elasticsearch 版本一致
 depends_on:
 elasticsearch:
 condition: service_healthy # 等待 Elasticsearch 健康后再啟動
 ports:
 - "5601:5601" # 將 Kibana 的 Web 界面端口映射到宿主機(jī)，通過 http://localhost:5601 訪問

docker compose -f /Users/cuihao/docker/EFK/docker-compose.yml up -d

=> ERROR [fluentd internal] load metadata for docker.io/fluent/fluentd:edge-debian 31.1s
 ------
 > [fluentd internal] load metadata for docker.io/fluent/fluentd:edge-debian:
 ------
 failed to solve: DeadlineExceeded: DeadlineExceeded: DeadlineExceeded: fluent/fluentd:edge-debian: failed to resolve source metadata for docker.io/fluent/fluentd:edge-debian: failed to authorize: DeadlineExceeded: failed to fetch anonymous token: Get "https://auth.docker.io/token?scope=repository%3Afluent%2Ffluentd%3Apull&service=registry.docker.io": dial tcp 75.126.124.162:443: i/o timeout

docker pull fluent/fluentd:edge-debian

docker compose -f /Users/cuihao/docker/EFK/docker-compose.yml up -d

##############################################
 # 輸入插件配置（收集日志）
 ##############################################

 # 輸入源1：采集cpu日志文件
 [INPUT]
 # 使用 tail 插件監(jiān)控文件變化
 Name tail
 # 自定義標(biāo)簽，標(biāo)識為系統(tǒng)指標(biāo)類日志
 Tag sys.metric 
 # 系統(tǒng)指標(biāo)日志文件路徑（可能是由其他工具生成的指標(biāo)數(shù)據(jù)）
 Path /var/log/metric_log
 # 檢查文件變化的間隔時間（秒）
 Refresh_Interval 10 

 ##############################################
 # 輸出插件（轉(zhuǎn)發(fā)到 Fluentd）
 ##############################################
 [OUTPUT]
 # 使用 forward 插件將日志轉(zhuǎn)發(fā)到Fluentd聚合器
 Name forward
 # 匹配所有標(biāo)簽的日志（* 是通配符，表示所有輸入源）
 Match *
 # Fluentd 服務(wù)地址（使用Docker Compose服務(wù)名進(jìn)行服務(wù)發(fā)現(xiàn)）
 Host fluentd
 # Fluentd 監(jiān)聽端口（forward插件的默認(rèn)端口）
 Port 24224
 # 網(wǎng)絡(luò)故障時的最大重試次數(shù)，防止無限重試消耗資源
 Retry_Limit 10

 # 輸出源2：同時輸出到控制臺（用于調(diào)試和監(jiān)控）
 [OUTPUT]
 # 使用 stdout 插件在控制臺打印日志
 Name stdout
 # 匹配所有標(biāo)簽的日志
 Match *
 # 注意：生產(chǎn)環(huán)境通常應(yīng)注釋或移除此輸出，避免日志重復(fù)和性能開銷

###############################
 # Fluentd 健康檢查（monitor_agent）
 ###############################
 <source>
 @type monitor_agent # 啟用監(jiān)控代理插件，用于收集Fluentd自身運行指標(biāo)
 bind 0.0.0.0 # 監(jiān)聽所有網(wǎng)絡(luò)接口
 port 24220 # 監(jiān)控服務(wù)端口，可通過此端口獲取Fluentd運行狀態(tài)信息
 </source>

 ###############################
 # Fluentd 日志輸入（forward）
 ###############################
 <source>
 @type forward # 啟用forward輸入插件，接收來自Fluent-bit或其他Fluentd節(jié)點的日志
 port 24224 # 轉(zhuǎn)發(fā)協(xié)議監(jiān)聽端口
 bind 0.0.0.0 # 監(jiān)聽所有網(wǎng)絡(luò)接口
 </source>

 ###############################
 # 處理 collectd 采集的CPU指標(biāo)數(shù)據(jù)
 ###############################
 # 第一步：提取log字段中write_log后的JSON部分（現(xiàn)有配置）
 <filter sys.metric> # 過濾系統(tǒng)指標(biāo)數(shù)據(jù)
 @type parser # 使用解析器插件
 key_name log # 解析log字段內(nèi)容
 reserve_data true # 保留所有原始字段
 remove_key_name_field false # 不移除原始的key_name字段
 <parse>
 @type regexp # 使用正則表達(dá)式解析
 # 正則表達(dá)式匹配collectd的CPU數(shù)據(jù)格式，使用命名捕獲組提取CPU信息
 expression /write_log values:(?:#012|\s)*(?<log>\[\{.*?\}\])/ # 非貪婪匹配確保只取一個數(shù)組

 time_key time # 指定時間字段
 time_type float # 時間格式為浮點數(shù)（Unix時間戳）
 </parse>
 </filter>

 # 第二步：將log字段的JSON字符串解析為JSON對象
 <filter sys.metric>
 @type parser
 key_name log # 解析第一步提取的log字段（此時是JSON字符串）
 reserve_data true # 保留所有字段
 remove_key_name_field false # 保留解析后的log字段
 <parse>
 @type json # 用JSON解析器處理
 json_parser json # 使用默認(rèn)JSON解析器
 array true # 強(qiáng)制將JSON數(shù)組字符串解析為數(shù)組對象
 </parse>
 </filter>

 # 第三步：標(biāo)記異常日志（增加詳細(xì)調(diào)試字段）
 <filter sys.metric>
 @type record_transformer
 enable_ruby true
 <record>
 # 直接判斷 values[0] 是否大于 80
 status ${(record["plugin"] == "cpu" && record["type"] == "percent" && record["values"].is_a?(Array) && !record["values"].empty? && record["values"][0].to_f > 80) ? "abnormal" : "normal"}

 debug_plugin ${record["plugin"].to_s}
 debug_type ${record["type"].to_s}
 debug_values_content ${record["values"].inspect}
 </record>
 </filter>

 ###############################
 # 輸出到 Elasticsearch 8、stdout 和 HTTP端點
 ###############################
 <match *.**> # 匹配所有標(biāo)簽的日志Tag
 @type copy # 復(fù)制插件，將日志同時發(fā)送到多個輸出目的地

 # 第一個輸出目標(biāo)：Elasticsearch
 <store>
 @id es_output
 @type elasticsearch # 輸出到Elasticsearch
 host elasticsearch # ES主機(jī)地址（可以是主機(jī)名或IP）
 port 9200 # ES服務(wù)端口
 scheme http # 使用HTTP協(xié)議
 logstash_format true # 使用Logstash格式索引命名
 logstash_prefix fluentd-${tag} # 索引前綴加上標(biāo)簽名
 logstash_dateformat %Y%m%d # 索引日期格式年月日
 include_tag_key true # 在輸出中包含標(biāo)簽字段
 tag_key @log_name # 標(biāo)簽字段的鍵名
 flush_interval 1s # 刷新間隔1秒
 </store>

 <store>
 @id output
 @type stdout # 輸出到控制臺/stdout
 key status # 字段
 pattern ^normal$ # 正則匹配
 </store>

 <store>
 @type relabel # 使用 relabel 輸出插件進(jìn)行標(biāo)簽重路由
 @label @abnormal # 將所有匹配到的數(shù)據(jù)重新路由到 @abnormal 標(biāo)簽處理流程
 </store>
 </match>


 ###############################
 # 異常日志輸出到 HTTP端點
 ###############################
 <label @abnormal>
 # 這個 filter 塊會對所有進(jìn)入 @abnormal 標(biāo)簽的數(shù)據(jù)進(jìn)行過濾
 # 只有通過這個過濾器的數(shù)據(jù)才會繼續(xù)流向后面的 match 塊
 <filter **>
 @type grep # 使用 grep 過濾器
 <regexp>
 key status # 檢查每條記錄的 status 字段
 pattern ^abnormal$ # 只保留 status 值為 "abnormal" 的記錄
 </regexp>
 </filter>

 # 這個 match 塊會接收到經(jīng)過上面 filter 過濾后的數(shù)據(jù)
 <match **>
 @id http_output
 @type http # HTTP輸出插件
 endpoint http://host.docker.internal:5001/analyze-fluentd-log # 日志分析API端點，host.docker.internal是Docker特殊主機(jī)名，指向宿主機(jī)
 http_method post # 使用POST方法發(fā)送

 <format>
 @type json # 數(shù)據(jù)格式為JSON
 </format>

 <buffer>
 # 緩沖配置區(qū)塊：控制數(shù)據(jù)如何緩沖和重試發(fā)送
 # 用于提高網(wǎng)絡(luò)輸出的可靠性和性能，避免頻繁的小數(shù)據(jù)包發(fā)送
 
 # 緩沖刷新間隔：每2秒強(qiáng)制刷新一次緩沖區(qū)
 # - 即使緩沖區(qū)未滿，也會每2秒發(fā)送一次累積的數(shù)據(jù)
 # - 平衡實時性和網(wǎng)絡(luò)效率：太短會增加網(wǎng)絡(luò)請求，太長會降低實時性
 flush_interval 2s
 
 # 重試策略：使用指數(shù)退避算法
 # - 第一次重試等待：基礎(chǔ)等待時間
 # - 第二次重試等待：基礎(chǔ)時間 × 2
 # - 第三次重試等待：基礎(chǔ)時間 × 4，依此類推
 # - 避免網(wǎng)絡(luò)擁塞時的大量重試導(dǎo)致雪崩效應(yīng)
 retry_type exponential_backoff
 
 # 基礎(chǔ)重試等待時間：第一次重試前等待1秒
 # - 首次重試的初始等待間隔
 # - 后續(xù)重試會根據(jù)指數(shù)退避算法遞增
 retry_wait 1s
 
 # 最大重試間隔：單次重試最多等待30秒
 # - 防止指數(shù)增長后的等待時間過長
 # - 即使使用指數(shù)退避，也不會超過30秒的間隔
 retry_max_interval 30s
 
 # 總重試超時時間：10分鐘后放棄重試
 # - 從第一次失敗開始計算，10分鐘后停止重試
 # - 防止因為長期不可用的目標(biāo)服務(wù)導(dǎo)致無限重試
 # - 超時后數(shù)據(jù)可能會被丟棄（取決于配置）
 retry_timeout 10m
 </buffer>
 </match>
 </label>

from flask import Flask, request, jsonify, Response
 from openai import AsyncOpenAI
 import os
 import json
 import threading
 import asyncio
 from datetime import datetime
 from dotenv import load_dotenv

 # 加載環(huán)境變量
 load_dotenv()

 # 創(chuàng)建Flask應(yīng)用實例
 app = Flask(__name__)

 # 初始化客戶端
 DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
 if not DEEPSEEK_API_KEY:
 raise ValueError("未找到DEEPSEEK_API_KEY，請在.env文件中配置")

 client = AsyncOpenAI(
 api_key=DEEPSEEK_API_KEY,
 base_url="https://api.deepseek.com"
 )

 # 路徑配置
 current_dir = os.path.dirname(os.path.abspath(__file__))
 report_dir = os.path.join(current_dir, "report")
 OUTPUT_REPORT_PATH = os.path.join(report_dir, "incident_report.json")
 ERROR_INFO_PATH = os.path.join(current_dir, "api_error.log")
 OPERATION_LOG_PATH = os.path.join(current_dir, "api_operation.log") # 新增操作日志文件
 os.makedirs(report_dir, exist_ok=True)


 def write_operation_log(message, request_id=None):
 """記錄操作日志，包含時間戳和可選的請求ID"""
 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] # 精確到毫秒
 request_id = request_id or "N/A"
 log_entry = f"[{timestamp}] [REQUEST={request_id}] OPERATION: {message}\n"
 try:
 with open(OPERATION_LOG_PATH, 'a', encoding='utf-8') as f:
 f.write(log_entry)
 print(log_entry.strip()) # 同時輸出到控制臺
 except Exception as e:
 print(f"[日志系統(tǒng)錯誤] 寫入操作日志失敗: {str(e)} | 原始消息: {message}")


 def write_error_log(error_message, request_id=None):
 """記錄錯誤日志，包含時間戳和請求ID"""
 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
 request_id = request_id or "N/A"
 log_entry = f"[{timestamp}] [REQUEST={request_id}] ERROR: {error_message}\n"
 try:
 with open(ERROR_INFO_PATH, 'a', encoding='utf-8') as f:
 f.write(log_entry)
 print(log_entry.strip()) # 同時輸出到控制臺
 except Exception as e:
 print(f"[日志系統(tǒng)錯誤] 寫入錯誤日志失敗: {str(e)} | 原始錯誤: {error_message}")


 def append_to_json_file(data, timestamp, request_id):
 """將數(shù)據(jù)追加到JSON文件，帶日志記錄"""
 try:
 write_operation_log("開始寫入分析報告", request_id)

 data_with_timestamp = {
 "timestamp": timestamp,
 "request_id": request_id, # 新增請求ID便于追蹤
 **data
 }

 existing_data = []
 if os.path.exists(OUTPUT_REPORT_PATH):
 try:
 with open(OUTPUT_REPORT_PATH, 'r', encoding='utf-8') as f:
 existing_data = json.load(f)
 write_operation_log(f"成功讀取現(xiàn)有報告（{len(existing_data)}條記錄）", request_id)
 except (json.JSONDecodeError, FileNotFoundError):
 error_msg = "報告文件損壞或不存在，將重新初始化"
 write_error_log(error_msg, request_id)
 existing_data = []

 existing_data.append(data_with_timestamp)

 with open(OUTPUT_REPORT_PATH, 'w', encoding='utf-8') as f:
 json.dump(existing_data, f, ensure_ascii=False, indent=2)

 write_operation_log("分析報告寫入成功", request_id)
 return True
 except Exception as e:
 error_msg = f"寫入JSON報告文件時出錯: {str(e)}"
 write_error_log(error_msg, request_id)
 return False


 async def process_llm_background(log_content, call_timestamp, request_id):
 """后臺處理LLM調(diào)用，帶詳細(xì)日志"""
 try:
 write_operation_log("開始后臺LLM處理流程", request_id)
 
 # 構(gòu)造提示詞
 write_operation_log("開始構(gòu)建提示詞", request_id)
 prompt = f"""
 你是一名系統(tǒng)運維專家。以下是異常日志，請生成根因分析報告：
 日志內(nèi)容：
 {log_content}

 請輸出格式：
 【標(biāo)題】
 ...
 【異常原因分析】
 ...
 【修復(fù)建議】
 ...
 """
 write_operation_log("提示詞構(gòu)建完成，準(zhǔn)備調(diào)用LLM", request_id)

 # 調(diào)用LLM
 write_operation_log("開始調(diào)用DeepSeek API", request_id)
 response = await client.chat.completions.create(
 model="deepseek-chat",
 messages=[
 {"role": "system", "content": "你是經(jīng)驗豐富的運維專家，負(fù)責(zé)分析日志并提供修復(fù)建議"},
 {"role": "user", "content": prompt}
 ],
 temperature=0.7,
 max_tokens=800,
 stream=False
 )
 write_operation_log("DeepSeek API調(diào)用成功", request_id)

 # 處理結(jié)果
 report_text = response.choices[0].message.content
 result_data = {
 "raw_log": log_content,
 "root_cause_report": report_text
 }
 write_operation_log("LLM返回結(jié)果解析完成", request_id)
 
 # 保存結(jié)果
 if append_to_json_file(result_data, call_timestamp, request_id):
 write_operation_log("分析結(jié)果已成功保存到報告文件", request_id)
 else:
 write_error_log("分析結(jié)果保存失敗", request_id)

 write_operation_log("后臺LLM處理流程完成", request_id)

 except Exception as e:
 error_msg = f"LLM調(diào)用失敗: {str(e)}"
 write_error_log(error_msg, request_id)


 def run_async_task(log_content, call_timestamp, request_id):
 """線程包裝器，帶日志"""
 try:
 write_operation_log("啟動異步任務(wù)處理線程", request_id)
 asyncio.run(process_llm_background(log_content, call_timestamp, request_id))
 except Exception as e:
 write_error_log(f"異步任務(wù)線程執(zhí)行失敗: {str(e)}", request_id)


 def build_log_content_from_records(records, request_id):
 """構(gòu)建日志內(nèi)容，帶日志記錄"""
 write_operation_log("開始構(gòu)建日志內(nèi)容", request_id)
 lines = []
 for i, record in enumerate(records or []):
 if isinstance(record, dict) and ("log" in record):
 value = record.get("log")
 if value is not None and str(value).strip() != "":
 lines.append(str(value))
 write_operation_log(f"成功提取第{i+1}條記錄的log字段", request_id)
 else:
 write_operation_log(f"第{i+1}條記錄的log字段為空，已忽略", request_id)
 else:
 write_operation_log(f"第{i+1}條記錄不包含log字段，已忽略", request_id)
 
 log_content = "\n".join(lines).strip()
 write_operation_log(f"日志內(nèi)容構(gòu)建完成，共包含{len(lines)}條有效記錄", request_id)
 return log_content


 @app.route('/analyze-fluentd-log', methods=['POST'])
 def analyze_fluentd_log():
 """主接口：帶詳細(xì)日志記錄的請求處理"""
 # 生成唯一請求ID，便于全流程追蹤
 request_id = datetime.now().strftime("%Y%m%d%H%M%S") + f"-{os.urandom(4).hex()}"
 write_operation_log("收到新的請求", request_id)

 try:
 # 1. 記錄請求基本信息
 call_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 write_operation_log(f"請求方法: {request.method}, 客戶端IP: {request.remote_addr}, Content-Type: {request.content_type}", request_id)

 # 2. 驗證請求類型
 if request.content_type != "application/x-ndjson":
 error_msg = f"不支持的Content-Type: {request.content_type}，僅接受application/x-ndjson"
 write_error_log(error_msg, request_id)
 return jsonify({
 "error": error_msg,
 "request_id": request_id
 }), 415

 # 3. 解析NDJSON數(shù)據(jù)
 write_operation_log("開始解析NDJSON數(shù)據(jù)", request_id)
 ndjson_data = request.data.decode().splitlines()
 records = []
 for i, line in enumerate(ndjson_data):
 if line.strip():
 try:
 record = json.loads(line)
 records.append(record)
 write_operation_log(f"成功解析第{i+1}行NDJSON數(shù)據(jù)", request_id)
 except json.JSONDecodeError as e:
 error_msg = f"第{i+1}行NDJSON解析失敗: {str(e)}"
 write_error_log(error_msg, request_id)
 records.append({"message": line, "parse_error": str(e)})
 else:
 write_operation_log(f"第{i+1}行是空白行，已忽略", request_id)
 write_operation_log(f"NDJSON數(shù)據(jù)解析完成，共{len(records)}條記錄", request_id)

 # 4. 過濾有效日志記錄
 write_operation_log("開始過濾有效日志記錄", request_id)
 log_records = [r for r in records if isinstance(r, dict) and 'log' in r and str(r.get('log', '')).strip()]
 write_operation_log(f"有效日志記錄過濾完成: {len(log_records)}/{len(records)}", request_id)
 
 if not log_records:
 error_msg = "未找到有效日志記錄（缺少log字段）"
 write_error_log(error_msg, request_id)
 return Response(
 json.dumps({
 "status": "ignored",
 "reason": error_msg,
 "raw_records_count": len(records),
 "request_id": request_id
 }, ensure_ascii=False, indent=2),
 mimetype="application/json"
 )

 # 5. 構(gòu)建日志內(nèi)容
 log_content = build_log_content_from_records(log_records, request_id)
 if not log_content:
 error_msg = "日志內(nèi)容為空"
 write_error_log(error_msg, request_id)
 return jsonify({
 "error": error_msg,
 "request_id": request_id
 }), 400

 # 6. 參數(shù)驗證通過，準(zhǔn)備啟動后臺任務(wù)
 write_operation_log("參數(shù)驗證通過，準(zhǔn)備啟動后臺處理任務(wù)", request_id)

 # 7. 啟動后臺線程處理LLM調(diào)用
 thread = threading.Thread(
 target=run_async_task,
 args=(log_content, call_timestamp, request_id),
 daemnotallow=True
 )
 thread.start()
 write_operation_log("后臺處理線程已啟動", request_id)

 # 8. 立即返回響應(yīng)
 response_data = {
 "status": "accepted",
 "message": "請求已接收，正在后臺處理",
 "timestamp": call_timestamp,
 "log_records_count": len(log_records),
 "request_id": request_id # 返回請求ID便于追蹤
 }
 write_operation_log("請求處理完成，已返回響應(yīng)", request_id)

 return Response(
 json.dumps(response_data, ensure_ascii=False, indent=2),
 mimetype="application/json",
 status=202
 )

 except Exception as e:
 error_msg = f"參數(shù)驗證失敗: {str(e)}"
 write_error_log(error_msg, request_id)
 return jsonify({
 "error": error_msg,
 "request_id": request_id
 }), 400


 @app.route('/health', methods=['GET'])
 def health_check():
 """健康檢查接口，帶日志"""
 request_id = f"health-{datetime.now().strftime('%Y%m%d%H%M%S')}"
 write_operation_log("收到健康檢查請求", request_id)
 response = jsonify({
 "status": "healthy",
 "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
 "request_id": request_id
 })
 write_operation_log("健康檢查響應(yīng)已返回", request_id)
 return response


 if __name__ == '__main__':
 print("Flask API 啟動中...")
 print("環(huán)境變量配置:")
 print(f" - 從 .env 文件加載 DEEPSEEK_API_KEY: {'已配置' if DEEPSEEK_API_KEY else '未配置'}")
 print("日志文件路徑:")
 print(f" - 操作日志: {OPERATION_LOG_PATH}")
 print(f" - 錯誤日志: {ERROR_INFO_PATH}")
 print("可用的端點:")
 print(" - POST /analyze-fluentd-log - 接收日志并在后臺處理")
 print(" - GET /health - 服務(wù)健康檢查")
 app.run(host="0.0.0.0", port=5001, debug=True)

import streamlit as st
 import json
 import re
 import os
 from log_generator import LogGenerator

 # 初始化日志生成器
 log_generator = LogGenerator()

 # ========================
 # 讀取 JSON 文件
 # ========================
 #file_path = r"report\incident_report.json"
 file_path = os.path.join("report", "incident_report.json")
 # 初始化reports變量
 reports = []

 # 檢查文件是否存在和讀取數(shù)據(jù)
 if os.path.exists(file_path) and os.path.getsize(file_path) > 0:
 try:
 with open(file_path, "r", encoding="utf-8") as f:
 reports = json.load(f)
 except json.JSONDecodeError as e:
 st.sidebar.error(f"JSON文件格式錯誤: {e}")
 except Exception as e:
 st.sidebar.error(f"讀取文件時發(fā)生錯誤: {e}")
 else:
 if not os.path.exists(file_path):
 st.sidebar.warning(f"?? 文件 {file_path} 不存在")
 else:
 st.sidebar.warning(f"?? 文件 {file_path} 為空")

 # ========================
 # 提取標(biāo)題（標(biāo)題在【標(biāo)題】后的下一行）
 # ========================
 def extract_title(report_text):
 # 去掉空行，逐行處理
 lines = [line.strip() for line in report_text.splitlines() if line.strip()]
 for i, line in enumerate(lines):
 if line.startswith("【標(biāo)題】"):
 # 找到【標(biāo)題】，取下一行作為真正的標(biāo)題
 if i + 1 < len(lines):
 return lines[i + 1].strip()
 else:
 return "未命名報告"
 return "未命名報告"

 # ========================
 # 按【字段】分段解析報告
 # 例如：分出 "異常原因分析"、"修復(fù)建議"、"異常日志"
 # ========================
 def split_sections(report_text):
 sections = {}
 # 用正則把【xxx】作為分隔符拆開
 parts = re.split(r"(【.*?】)", report_text)
 current_key = None
 for part in parts:
 if not part.strip():
 continue
 if part.startswith("【") and part.endswith("】"):
 # 當(dāng)前是一個小節(jié)標(biāo)題，例如【異常原因分析】
 current_key = part.strip("【】")
 sections[current_key] = ""
 else:
 # 當(dāng)前是小節(jié)內(nèi)容
 if current_key:
 sections[current_key] += part.strip() + "\n"
 return sections

 # ========================
 # 側(cè)邊欄：搜索功能
 # ========================
 st.sidebar.title("異常報告列表")

 # 如果沒有報告數(shù)據(jù)，顯示提示信息
 if not reports:
 st.sidebar.info("?? 暫無報告數(shù)據(jù)")
 # 右側(cè)主內(nèi)容區(qū)域顯示友好提示
 st.title("?? 異常報告系統(tǒng)")
 st.info("""
 ## 歡迎使用異常報告系統(tǒng)！
 
 ### ?? 當(dāng)前狀態(tài)
 - 報告數(shù)據(jù)文件未找到或為空
 - 請確保 `report/incident_report.json` 文件存在且包含有效數(shù)據(jù)
 
 ### ?? 如何添加數(shù)據(jù)
 1. 在 `report` 文件夾中創(chuàng)建 `incident_report.json` 文件
 2. 按照以下格式添加報告數(shù)據(jù)：
 ```json
 [
 {
 "timestamp": "2024-01-01 10:00:00",
 "raw_log": "日志內(nèi)容...",
 "root_cause_report": "【標(biāo)題】\n報告標(biāo)題\n【異常原因分析】\n分析內(nèi)容..."
 }
 ]
 ```
 
 ### ?? 功能說明
 - 支持按時間倒序查看報告
 - 支持搜索報告內(nèi)容
 - 自動解析報告結(jié)構(gòu)并展示
 """)
 st.stop()

 # 倒序排序：最新的報告排在最前面
 reports_sorted = sorted(reports, key=lambda x: x["timestamp"], reverse=True)

 # 添加搜索輸入框
 search_query = st.sidebar.text_input("?? 搜索報告內(nèi)容（標(biāo)題/日志/分析/建議）").strip()

 # 判斷一條報告是否匹配搜索條件
 def report_matches(report, query):
 if not query:
 return True
 query = query.lower()
 if query in extract_title(report["root_cause_report"]).lower():
 return True
 if query in report["raw_log"].lower():
 return True
 if query in report["root_cause_report"].lower():
 return True
 return False

 # 根據(jù)搜索條件過濾報告
 filtered_reports = [r for r in reports_sorted if report_matches(r, search_query)]

 # 構(gòu)建側(cè)邊欄顯示內(nèi)容（報告生成時間 | 標(biāo)題）
 sidebar_items = [
 f"{r['timestamp']} | {extract_title(r['root_cause_report'])}" for r in filtered_reports
 ]

 # 沒有結(jié)果時提示
 if not sidebar_items:
 st.sidebar.warning("未找到匹配的報告")
 st.stop()

 # 側(cè)邊欄選擇框
 selected_item = st.sidebar.radio("選擇報告", sidebar_items)

 # ========================
 # 側(cè)邊欄：測試區(qū)域
 # ========================
 st.sidebar.markdown("---")
 st.sidebar.subheader("?? 測試數(shù)據(jù)生成")

 # 日志類型選擇
 log_type = st.sidebar.radio(
 "選擇日志類型",
 ["CPU"],
 help="CPU: 生成CPU使用率日志",
 )

 # CPU使用率輸入（僅在選擇CPU類型時顯示）
 cpu_usage = None
 if log_type == "CPU":
 cpu_usage = st.sidebar.number_input(
 "CPU使用率 (%)",
 min_value=0.1,
 max_value=100.0,
 value=85.0,
 step=0.1,
 help="輸入0.1-100之間的CPU使用率值"
 )

 # 生成按鈕
 if st.sidebar.button(f"生成{log_type}日志", type="primary"):
 with st.spinner("正在生成日志..."):
 result = log_generator.generate_and_save_log(log_type.lower(), cpu_usage)
 
 if result['status'] == 'success':
 st.sidebar.success(result['message'])
 st.sidebar.info(f"日志文件: {result['log_file']}")
 if 'log_content_preview' in result:
 st.sidebar.code(result['log_content_preview'], language="text")
 else:
 st.sidebar.error(result['message'])

 # 根據(jù)選擇找到對應(yīng)的報告
 selected_index = sidebar_items.index(selected_item)
 selected_report = filtered_reports[selected_index]

 # 提取報告標(biāo)題
 report_title = extract_title(selected_report["root_cause_report"])

 # ========================
 # 右側(cè)內(nèi)容展示
 # ========================
 st.title(report_title)
 st.subheader(f"**報告生成時間**: {selected_report['timestamp']}")

 # ---- 原始日志（默認(rèn)折疊） ----
 with st.expander("原始日志", expanded=False):
 st.code(selected_report["raw_log"], language="text")

 # ---- 根因分析報告 ----
 sections = split_sections(selected_report["root_cause_report"])
 # 去掉【標(biāo)題】部分，避免重復(fù)展示
 sections.pop("標(biāo)題", None)


 for key, value in sections.items():
 st.subheader(key) # 小節(jié)標(biāo)題，例如 “異常原因分析” st.markdown(value.strip()) # 小節(jié)內(nèi)容

python process_log.py

Flask API 啟動中...
 環(huán)境變量配置:
 - 從 .env 文件加載 DEEPSEEK_API_KEY: 已配置
 日志文件路徑:
 - 操作日志: /Users/cuihao/docker/EFK/api_operation.log
 - 錯誤日志: /Users/cuihao/docker/EFK/api_error.log
 可用的端點:
 - POST /analyze-fluentd-log - 接收日志并在后臺處理
 - GET /health - 服務(wù)健康檢查
 * Serving Flask app "process_log" (lazy loading)

streamlit run log_ui.py