b3012b8320
- ai-service: Ollama 기반 AI 서비스 (분류, 시맨틱 검색, RAG Q&A, 패턴 분석) - AI 어시스턴트 페이지: 채팅형 Q&A, 시맨틱 검색, 패턴 분석, 분류 테스트 - 권한 시스템에 ai_assistant 페이지 등록 (기본 비활성) - 기존 페이지에 AI 기능 통합 (대시보드, 수신함, 관리함) - docker-compose, gateway, nginx 설정 업데이트 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
from urllib.parse import quote_plus
|
|
|
|
from sqlalchemy import create_engine, text
|
|
from config import settings
|
|
|
|
|
|
def get_engine():
|
|
password = quote_plus(settings.DB_PASSWORD)
|
|
url = (
|
|
f"mysql+pymysql://{settings.DB_USER}:{password}"
|
|
f"@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}"
|
|
)
|
|
return create_engine(url, pool_pre_ping=True, pool_size=5)
|
|
|
|
|
|
engine = get_engine()
|
|
|
|
|
|
def get_all_issues() -> list[dict]:
|
|
with engine.connect() as conn:
|
|
result = conn.execute(
|
|
text(
|
|
"SELECT id, category, description, detail_notes, "
|
|
"final_description, final_category, solution, "
|
|
"management_comment, cause_detail, project_id, "
|
|
"review_status, report_date, responsible_department, "
|
|
"location_info "
|
|
"FROM qc_issues ORDER BY id"
|
|
)
|
|
)
|
|
return [dict(row._mapping) for row in result]
|
|
|
|
|
|
def get_issue_by_id(issue_id: int) -> dict | None:
|
|
with engine.connect() as conn:
|
|
result = conn.execute(
|
|
text(
|
|
"SELECT id, category, description, detail_notes, "
|
|
"final_description, final_category, solution, "
|
|
"management_comment, cause_detail, project_id, "
|
|
"review_status, report_date, responsible_department, "
|
|
"location_info "
|
|
"FROM qc_issues WHERE id = :id"
|
|
),
|
|
{"id": issue_id},
|
|
)
|
|
row = result.fetchone()
|
|
return dict(row._mapping) if row else None
|
|
|
|
|
|
def get_issues_since(last_id: int) -> list[dict]:
|
|
with engine.connect() as conn:
|
|
result = conn.execute(
|
|
text(
|
|
"SELECT id, category, description, detail_notes, "
|
|
"final_description, final_category, solution, "
|
|
"management_comment, cause_detail, project_id, "
|
|
"review_status, report_date, responsible_department, "
|
|
"location_info "
|
|
"FROM qc_issues WHERE id > :last_id ORDER BY id"
|
|
),
|
|
{"last_id": last_id},
|
|
)
|
|
return [dict(row._mapping) for row in result]
|
|
|
|
|
|
def get_daily_qc_stats(date_str: str) -> dict:
|
|
with engine.connect() as conn:
|
|
result = conn.execute(
|
|
text(
|
|
"SELECT "
|
|
" COUNT(*) as total, "
|
|
" SUM(CASE WHEN DATE(report_date) = :d THEN 1 ELSE 0 END) as new_today, "
|
|
" SUM(CASE WHEN review_status = 'in_progress' THEN 1 ELSE 0 END) as in_progress, "
|
|
" SUM(CASE WHEN review_status = 'completed' THEN 1 ELSE 0 END) as completed, "
|
|
" SUM(CASE WHEN review_status = 'pending_review' THEN 1 ELSE 0 END) as pending "
|
|
"FROM qc_issues"
|
|
),
|
|
{"d": date_str},
|
|
)
|
|
row = result.fetchone()
|
|
return dict(row._mapping) if row else {}
|
|
|
|
|
|
def get_issues_for_date(date_str: str) -> list[dict]:
|
|
with engine.connect() as conn:
|
|
result = conn.execute(
|
|
text(
|
|
"SELECT id, category, description, detail_notes, "
|
|
"review_status, responsible_department, solution "
|
|
"FROM qc_issues "
|
|
"WHERE DATE(report_date) = :d "
|
|
"ORDER BY id"
|
|
),
|
|
{"d": date_str},
|
|
)
|
|
return [dict(row._mapping) for row in result]
|