import google.generativeai as genai
import yfinance as yf
import requests, json, logging, re, os, time, sqlite3, feedparser, io, base64
import shutil
from datetime import datetime
import pytz
from PIL import Image
from config import settings
from prompts.news_gemini_prompts import GEMINI_NEWS_PROMPT
class NewsAnalyzerGemini:
def __init__(self):
"""
[V285.0-FINAL] BridgeMatrix Lab 자율형 에이전트
- 엔진: Gemini 2.5 Flash (추론 비용 0원)
- 지표: 7대 핵심 지표 수집 및 차트 날짜(%m/%d) 라벨링 자동화
- 이미지: Imagen 4.0 최신 모델 및 소유권(nginx) 자동 이양
- 로그: GPT 관련 불필요 메시지 삭제 및 정제
"""
genai.configure(api_key=settings.GOOGLE_API_KEY)
model_id = 'gemini-2.5-flash'
# 🚩 [V285.0] 불필요 로그 삭제 및 Gemini 엔진 시동 확인
try:
self.model = genai.GenerativeModel(model_name=model_id, tools=['google_search'])
self.model.generate_content("ping", generation_config={"max_output_tokens": 1})
logging.info(f"🚀 BridgeMatrix Lab: Gemini 2.5 Flash 분석 엔진 시동 완료")
except:
self.model = genai.GenerativeModel(model_name=model_id)
logging.warning("🚩 [Safe Mode] 도구 없이 Gemini 2.5 Flash 단독 가동")
self.KST = pytz.timezone('Asia/Seoul')
self.db_path = os.path.join(settings.BASE_DIR, 'database', 'bridgematrix_lab.db')
self.image_dir = "/var/www/html/images"
self.web_base_url = "https://demoportal.kr/images"
# 폴더 자동 생성 및 권한 설정
if not os.path.exists(self.image_dir):
os.makedirs(self.image_dir, mode=0o755, exist_ok=True)
try: shutil.chown(self.image_dir, user="nginx", group="nginx")
except: pass
self._init_db()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("CREATE TABLE IF NOT EXISTS processed_news (link TEXT PRIMARY KEY, title TEXT, processed_at DATETIME DEFAULT CURRENT_TIMESTAMP)")
def analyze_news(self, news_item):
"""Gemini 2.5 Flash 기반 3,500자 분석 및 7대 지표 통합 리포트 생성"""
raw_title = news_item['title'].replace("특이주", "특징주")
try:
logging.info(f"🔍 단계 1: 제미나이 분석 엔진 가동 ({raw_title})")
response = self.model.generate_content(f"분석 대상 속보: {raw_title}\n\n{GEMINI_NEWS_PROMPT}")
match = re.search(r'\{.*\}', response.text, re.DOTALL)
data = json.loads(match.group()) if match else {"analysis_ko": response.text}
logging.info("🎨 단계 2: Imagen 4.0 기반 맞춤형 배너 생성 중...")
thumb_url = self._generate_optimized_banner(raw_title)
return {
"headline": f"[{datetime.now(self.KST).strftime('%m/%d 특징주')}] {raw_title}",
"analysis_ko": self._format_analysis(data.get("analysis_ko", "")),
"market_data": self._get_market_intelligence(),
"summary_3_lines": data.get("summary_3_lines", ""),
"sector_strategy": data.get("sector_strategy", []),
"formatted_date": datetime.now(self.KST).strftime("%Y-%m-%d %H:%M"),
"thumbnail_url": thumb_url
}
except Exception as e:
logging.error(f"❌ 분석 오류 발생: {e}"); return None
def _get_market_intelligence(self):
"""📈 7대 핵심 지표 수집 및 차트 날짜 라벨(%m/%d) 동적 생성"""
intel = {
"indices": {}, "korea": {}, "global": {},
"labels": [] # 🚩 차트 X축 날짜 라벨
}
target_map = {
"KOSPI": "^KS11", "KOSDAQ": "^KQ11", "NASDAQ": "^IXIC",
"S&P500": "^GSPC", "USD/KRW": "USDKRW=X", "WTI유가": "CL=F",
"필라델피아반도체": "^SOX"
}
for name, ticker in target_map.items():
try:
tk = yf.Ticker(ticker)
# 7일치를 가져와 주말 제외 최근 5거래일 추출
data = tk.history(period="7d").tail(5)
if not data.empty:
# 🚩 [V285.0 핵심] 차트 라벨을 '3/22' 형식의 실제 날짜로 구성
if name == "KOSPI" and not intel["labels"]:
intel["labels"] = [d.strftime('%m/%d') for d in data.index]
curr, prev = data['Close'].iloc[-1], data['Close'].iloc[-2]
rate = ((curr / prev) - 1) * 100
item_data = {
"val": f"{curr:,.2f}",
"rate": f"{rate:+.2f}%",
"trend": "up" if rate > 0 else "down",
"bottom_color": "#dc2626" if rate > 0 else "#2563eb",
"history": [round(x, 2) for x in data['Close'].tolist()]
}
intel["indices"][name] = item_data
if name in ["KOSPI", "KOSDAQ", "USD/KRW"]:
intel["korea"][name] = item_data
else:
intel["global"][name] = item_data
except Exception as e:
logging.error(f"지표 수집 실패 ({name}): {e}")
if not intel["labels"]: intel["labels"] = ["-", "-", "-", "-", "Now"]
return intel
def _generate_optimized_banner(self, headline):
"""Imagen 4.0 배너 생성 및 소유권(nginx) 강제 이양"""
default_img = f"{self.web_base_url}/default_news.webp"
# 🚩 [V285.0] Imagen 4.0 엔드포인트 적용
imagen_model = "imagen-4.0-generate-001"
url = f"https://generativelanguage.googleapis.com/v1beta/models/{imagen_model}:predict?key={settings.GOOGLE_API_KEY}"
try:
payload = {"instances": [{"prompt": f"Professional financial 3D conceptual art for: {headline}, dark navy style"}]}
res = requests.post(url, json=payload, timeout=40)
res_data = res.json()
img_b64 = None
if 'predictions' in res_data and len(res_data['predictions']) > 0:
img_b64 = res_data['predictions'][0].get('bytesBase64Encoded')
elif 'images' in res_data and len(res_data['images']) > 0:
img_b64 = res_data['images'][0].get('bytesBase64Encoded')
if img_b64:
fname = f"bml_{int(time.time())}.webp"
fpath = os.path.join(self.image_dir, fname)
img = Image.open(io.BytesIO(base64.b64decode(img_b64))).convert("RGB")
img.thumbnail((1200, 675))
img.save(fpath, "WEBP", quality=85)
# 🚩 Nginx 읽기 권한 및 소유권 즉시 설정
os.chmod(fpath, 0o644)
try: shutil.chown(fpath, user="nginx", group="nginx")
except: pass
logging.info(f"✅ 배너 생성 완료 및 권한 설정 성공: {fname}")
return f"{self.web_base_url}/{fname}"
return default_img
except: return default_img
def _format_analysis(self, text):
"""가독성 최적화: 볼드체 처리 및 섹션 디자인 적용"""
if not text: return ""
summary_bar = f"
⚡ Quick View: 실시간 지표와 AI 통찰을 결합한 투자 전략입니다.
"
# 소제목 📍 스타일링
text = re.sub(r'📍([^<>\n]+)', r'
\1
', text)
paragraphs = text.split('\n')
html_list = []
for p in paragraphs:
p = p.strip()
if not p or "
\1', p)
html_list.append(f"{p}
")
return summary_bar + "".join(html_list)