| 날짜 | 방문 | AI상담 | 이미지 | 파일 | 오피스 |
|---|
VIDraft/JGOS-stats · 15요청마다 자동 저장 · KST 기준
" "" ) @app.get("/api/greeting") async def greeting(): return {"greeting": GREETING} @app.post("/api/image") async def gen_image(request: Request): try: bump("image_gen") except Exception: pass body = await request.json() prompt = (body.get("prompt") or body.get("message") or "").strip() if not prompt: return JSONResponse({"error": "프롬프트가 비었습니다."}, status_code=400) fal_key = os.getenv("FAL_KEY", "") if not fal_key: return JSONResponse({"error": "이미지 생성 키(FAL_KEY)가 설정되지 않았습니다."}, status_code=503) payload = { "prompt": prompt, "num_images": int(body.get("num_images", 1) or 1), "aspect_ratio": body.get("aspect_ratio", "1:1"), "resolution": body.get("resolution", "1k"), "output_format": "png", } try: async with httpx.AsyncClient(timeout=180) as client: r = await client.post( "https://fal.run/xai/grok-imagine-image", headers={"Authorization": "Key " + fal_key, "Content-Type": "application/json"}, json=payload) if r.status_code != 200: return JSONResponse({"error": "이미지 생성 실패 (%d)" % r.status_code}, status_code=502) data = r.json() imgs = data.get("images") or [] url = imgs[0].get("url") if imgs else None if not url: return JSONResponse({"error": "이미지 URL을 받지 못했습니다."}, status_code=502) ir = await client.get(url) b64 = base64.b64encode(ir.content).decode("ascii") ct = ir.headers.get("content-type", "image/png") return {"image": "data:%s;base64,%s" % (ct, b64), "url": url} except Exception as e: return JSONResponse({"error": "이미지 생성 오류: %s" % e}, status_code=502) async def gen_image(request: Request): """JGOS-Image — Qwen-Image-2512 (B200 cl1 GPU1) 한글 이미지 생성 프록시. JGOS_IMAGE_API_URL 미설정 시 503 '준비 중' 반환 (탭은 보이되 생성 보류).""" body = await request.json() prompt = (body.get("prompt") or "").strip() if not prompt: return JSONResponse({"error": "프롬프트를 입력하세요."}, status_code=400) if not IMAGE_API_URL: return JSONResponse({"error": "JGOS-Image 생성 서버를 준비 중입니다. 곧 제공됩니다."}, status_code=503) payload = { "prompt": prompt, "width": int(body.get("width", 1024)), "height": int(body.get("height", 1024)), "steps": int(body.get("steps", 30)), } try: async with httpx.AsyncClient(timeout=180) as client: r = await client.post( f"{IMAGE_API_URL}/generate", headers={"Authorization": f"Bearer {IMAGE_API_KEY}", "Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (JGOS-Image)"}, json=payload, ) if r.status_code != 200: return JSONResponse({"error": f"이미지 서버 오류 HTTP {r.status_code}"}, status_code=502) data = r.json() img = data.get("image") or data.get("image_url") or data.get("data") if not img: return JSONResponse({"error": "이미지 응답 형식 오류"}, status_code=502) return {"image": img} except httpx.ConnectError: return JSONResponse({"error": "이미지 생성 서버 연결 실패 (준비 중)."}, status_code=503) except Exception as e: return JSONResponse({"error": f"{type(e).__name__}: {str(e)[:150]}"}, status_code=502) # ====== 파일 변환 함수들 ====== def cap(text: str, n: int = MAX_TEXT_CHARS) -> str: if len(text) <= n: return text return text[:n] + f"\n\n[... 이후 {len(text)-n:,}자 생략 (총 {len(text):,}자)]" def extract_pdf(data: bytes) -> str: import fitz doc = fitz.open(stream=data, filetype="pdf") parts = [] for page in doc: t = page.get_text("text") if t: parts.append(t) doc.close() return "\n\n".join(parts) def extract_docx(data: bytes) -> str: from docx import Document doc = Document(io.BytesIO(data)) out = [] for p in doc.paragraphs: if p.text.strip(): out.append(p.text) for tbl in doc.tables: for row in tbl.rows: out.append(" | ".join(cell.text.strip() for cell in row.cells)) return "\n".join(out) def extract_pptx(data: bytes) -> str: from pptx import Presentation prs = Presentation(io.BytesIO(data)) out = [] for i, slide in enumerate(prs.slides, 1): out.append(f"\n=== Slide {i} ===") for shape in slide.shapes: if shape.has_text_frame: for p in shape.text_frame.paragraphs: text = "".join(r.text for r in p.runs) if text.strip(): out.append(text) if shape.has_table: for row in shape.table.rows: out.append(" | ".join(c.text.strip() for c in row.cells)) return "\n".join(out) def extract_xlsx(data: bytes) -> str: import openpyxl wb = openpyxl.load_workbook(io.BytesIO(data), data_only=True, read_only=True) out = [] for sn in wb.sheetnames: ws = wb[sn] out.append(f"\n=== Sheet: {sn} ===") for row in ws.iter_rows(values_only=True): cells = [str(c) if c is not None else "" for c in row] if any(cells): out.append(" | ".join(cells)) wb.close() return "\n".join(out) def extract_xls(data: bytes) -> str: import xlrd wb = xlrd.open_workbook(file_contents=data) out = [] for sn in wb.sheet_names(): ws = wb.sheet_by_name(sn) out.append(f"\n=== Sheet: {sn} ===") for r in range(ws.nrows): row = [str(ws.cell_value(r, c)) for c in range(ws.ncols)] if any(row): out.append(" | ".join(row)) return "\n".join(out) def extract_csv(data: bytes) -> str: import pandas as pd for enc in ("utf-8", "cp949", "euc-kr", "latin1"): try: df = pd.read_csv(io.BytesIO(data), encoding=enc) return df.to_markdown(index=False) if len(df) <= 500 else df.head(500).to_markdown(index=False) + f"\n\n[... 총 {len(df):,}행 중 500행 표시]" except UnicodeDecodeError: continue except Exception as e: return f"[CSV 읽기 실패: {e}]" return "[CSV 인코딩 식별 실패]" def extract_txt(data: bytes) -> str: for enc in ("utf-8", "cp949", "euc-kr", "latin1"): try: return data.decode(enc) except UnicodeDecodeError: continue return data.decode("utf-8", errors="replace") import tempfile, asyncio as _asyncio PKG_DIR = Path(__file__).parent / "pkg" RHWP_SCRIPT = PKG_DIR / "extract_hwp.mjs" async def extract_hwp(data: bytes, fname: str) -> str: """HanSoo rhwp WASM (Rust) — Node.js subprocess로 HWP/HWPX 한글 텍스트 직접 추출. 외부 API 의존성 0. JGOS Space 내부 처리. """ if not RHWP_SCRIPT.exists(): return f"[HWP 처리기 미설치: {RHWP_SCRIPT}]" suffix = ".hwpx" if fname.lower().endswith(".hwpx") else ".hwp" tmp_path = None try: with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tf: tf.write(data); tf.flush(); tmp_path = tf.name proc = await _asyncio.create_subprocess_exec( "node", str(RHWP_SCRIPT), tmp_path, stdout=_asyncio.subprocess.PIPE, stderr=_asyncio.subprocess.PIPE, ) try: stdout, stderr = await _asyncio.wait_for(proc.communicate(), timeout=60) except _asyncio.TimeoutError: proc.kill(); return f"[HWP 변환 타임아웃 60s: {fname}]" if proc.returncode != 0: err = (stderr.decode("utf-8", errors="ignore")[:300]) if stderr else "no stderr" return f"[HWP 변환 실패 (exit {proc.returncode}): {err}]" try: result = json.loads(stdout.decode("utf-8", errors="ignore")) except Exception as e: return f"[HWP 결과 파싱 실패: {e}]" if not result.get("ok"): return f"[HWP 추출 실패: {result.get('error','unknown')}]" text = result.get("text", "") return text or f"[{fname}: 추출 텍스트 없음 (paragraphs={result.get('paragraphs',0)})]" except FileNotFoundError as e: return f"[Node.js 미설치 — Docker rebuild 필요: {e}]" except Exception as e: return f"[HWP 처리 오류: {type(e).__name__}: {e}]" finally: if tmp_path: try: os.unlink(tmp_path) except: pass def extract_image_meta(data: bytes, fname: str) -> str: """이미지: 메타만 텍스트로. 실제 vision 분석은 base64로 LLM에 직접 전달.""" try: from PIL import Image img = Image.open(io.BytesIO(data)) return f"[이미지: {fname} · {img.format} · {img.size[0]}×{img.size[1]}px · LLM에 직접 전달됨]" except Exception as e: return f"[이미지 메타 추출 실패: {e}]" IMAGE_EXTS = {"jpg", "jpeg", "png", "gif", "webp"} IMAGE_MIME = {"jpg": "jpeg", "jpeg": "jpeg", "png": "png", "gif": "gif", "webp": "webp"} @app.post("/api/upload") async def upload(file: UploadFile = File(...)): name = file.filename or "untitled" ext = name.rsplit(".", 1)[-1].lower() if "." in name else "" data = await file.read() size_mb = len(data) / 1024 / 1024 if size_mb > MAX_UPLOAD_MB: raise HTTPException(413, f"파일 너무 큼 ({size_mb:.1f}MB > {MAX_UPLOAD_MB}MB)") image_data_url = None try: if ext == "pdf": text = extract_pdf(data) elif ext == "docx": text = extract_docx(data) elif ext == "pptx": text = extract_pptx(data) elif ext == "xlsx": text = extract_xlsx(data) elif ext == "xls": text = extract_xls(data) elif ext == "csv": text = extract_csv(data) elif ext in ("txt", "md"): text = extract_txt(data) elif ext in ("hwp", "hwpx"): text = await extract_hwp(data, name) elif ext in IMAGE_EXTS: text = extract_image_meta(data, name) # ⭐ 이미지는 base64 data URL로 LLM에 직접 전달 mime = IMAGE_MIME.get(ext, "jpeg") image_data_url = f"data:image/{mime};base64,{base64.b64encode(data).decode('ascii')}" else: raise HTTPException(415, f"지원하지 않는 파일 형식: .{ext}") except HTTPException: raise except Exception as e: return JSONResponse({"error": f"{type(e).__name__}: {e}", "text": ""}, status_code=500) capped = cap(text or "") resp = { "ok": True, "filename": name, "ext": ext, "size_kb": round(len(data) / 1024, 1), "text_len_raw": len(text or ""), "text_len_returned": len(capped), "text": capped, "meta": {"capped": len(text or "") > MAX_TEXT_CHARS}, } if image_data_url: resp["image_data_url"] = image_data_url resp["is_image"] = True return resp # ===== 웹 검색 (네이버 1차 + Brave 2차) — 시민 질의 실시간 grounding ===== import re as _re_ws _TAG_RE = _re_ws.compile(r"<[^>]+>") def _clean_html(s: str) -> str: s = _TAG_RE.sub("", s or "") for a, b in ((""", '"'), ("&", "&"), ("<", "<"), (">", ">"), ("'", "'")): s = s.replace(a, b) return s.strip() _KW_LOCAL = ("어디", "주소", "위치", "찾아", "가는길", "가는 길", "근처", "오시는", "약도", "연락처", "전화번호") _KW_NEWS = ("뉴스", "최근", "오늘", "현재", "근황", "요즘", "최신", "속보", "이번주", "이번 주", "어제", "지금") _KW_BLOG = ("맛집", "여행", "가볼만", "가 볼만", "후기", "축제", "숙소", "카페", "코스", "데이트", "당일치기", "1박2일") _SRC_LABEL = {"local": "네이버 지역", "news": "네이버 뉴스", "blog": "네이버 블로그", "webkr": "네이버 웹", "encyc": "지식백과"} async def _naver(query: str, kind: str, n: int = 4, sort: str = None) -> list: if not (NAVER_CLIENT_ID and NAVER_CLIENT_SECRET): return [] params = {"query": query, "display": n} if sort: params["sort"] = sort try: async with httpx.AsyncClient(timeout=8) as c: r = await c.get(f"https://openapi.naver.com/v1/search/{kind}.json", params=params, headers={"X-Naver-Client-Id": NAVER_CLIENT_ID, "X-Naver-Client-Secret": NAVER_CLIENT_SECRET}) if r.status_code != 200: return [] def _fmt_date(it): pd = it.get("pubDate") or it.get("postdate") or "" if not pd: return "" # postdate: 20220926 → 2022-09-26 if pd.isdigit() and len(pd) == 8: return f"{pd[:4]}-{pd[4:6]}-{pd[6:]}" return pd return [{ "title": _clean_html(it.get("title")), "desc": _clean_html(it.get("description")), "link": it.get("link", ""), "addr": it.get("roadAddress") or it.get("address") or "", "src": _SRC_LABEL.get(kind, "네이버"), "date": _fmt_date(it), } for it in r.json().get("items", [])] except Exception: return [] async def _brave(query: str, n: int = 4) -> list: if not BRAVE_API_KEY: return [] try: async with httpx.AsyncClient(timeout=8) as c: r = await c.get("https://api.search.brave.com/res/v1/web/search", params={"q": query, "count": n, "country": "kr", "search_lang": "ko"}, headers={"X-Subscription-Token": BRAVE_API_KEY, "Accept": "application/json"}) if r.status_code != 200: return [] return [{ "title": _clean_html(it.get("title")), "desc": _clean_html(it.get("description")), "link": it.get("url", ""), "addr": "", "src": "Brave", } for it in r.json().get("web", {}).get("results", [])] except Exception: return [] def _is_fresh_query(q: str) -> bool: """토글 OFF여도 자동 검색을 유발하는 '최신성' 키워드.""" return any(k in q for k in _KW_NEWS) async def run_web_search(query: str) -> list: q_raw = (query or "").strip() if not q_raw: return [] q = _clean_query(q_raw) # 검색어 정제 (조사·명령어 제거 → 고유명사 위주) # webkr(권위 웹문서) + news(최신순)는 항상 함께 — 최신성 보장 jobs = [_naver(q, "webkr", 4), _naver(q, "news", 5, sort="date")] if any(k in q_raw for k in _KW_LOCAL): jobs.append(_naver(q, "local", 4)) if any(k in q_raw for k in _KW_BLOG): jobs.append(_naver(q, "blog", 4)) groups = await asyncio.gather(*jobs) results, seen = [], set() for g in groups: for it in g: key = it["link"] or it["title"] if key and key not in seen: seen.add(key) results.append(it) # 날짜 있는 항목(뉴스 등)을 최신순으로 앞에, 날짜 없는 권위문서는 뒤에 def _date_key(it): import re d = it.get("date", "") # YYYY-MM-DD 또는 RFC 형식 → 정렬용 문자열 (없으면 최하위) m = re.search(r"(\d{4})[-.]?(\d{2})[-.]?(\d{2})", d) return m.group(0).replace(".", "-") if m else "0000-00-00" results.sort(key=_date_key, reverse=True) if len(results) < 3: # 네이버 빈약 → Brave 글로벌 보조 for it in await _brave(q, 4): key = it["link"] or it["title"] if key and key not in seen: seen.add(key) results.append(it) return results[:10] def build_search_context(results: list) -> str: head = ("【실시간 웹 검색 결과 — 이것이 정답의 근거입니다】 아래는 방금 조회한 최신 정보(최신순 정렬)입니다. " "★당신의 사전 학습 지식보다 아래 검색 결과가 절대적으로 우선합니다. 당신이 알던 것과 다르면 '아래 검색 결과가 맞습니다'. " "반드시 아래 내용만을 근거로 답하고, 관련 문장 끝에 [1][2]처럼 출처 번호를 다세요. " "검색 결과에 명시되지 않은 내용은 절대 지어내지 말고 '검색에서 확인되지 않았다'고 솔직히 말하세요. " "여러 항목이 충돌하면 '날짜'가 최근인 항목을 진실로 채택하세요.") lines = [head] for i, it in enumerate(results, 1): seg = f"[{i}] ({it['src']}) {it['title']}" if it.get("date"): seg += f" · 날짜: {it['date']}" if it.get("addr"): seg += f" · 주소: {it['addr']}" if it.get("desc"): seg += f" — {it['desc']}" lines.append(seg) return "\n".join(lines) _HANGUL_RE = _re_ws.compile(r'[가-힣]') def _find_body_start(raw: str): """reasoning을 건너뛰고 본문 시작 인덱스 반환 (없으면 None).""" # ★최우선: '최종 답변' 마커가 있으면 그 이후가 본문 (reasoning 서술형/라벨형 무관) m_final = _re_ws.search(r'(?:최종\s*답변|최종답변|\[\s*최종\s*답변\s*\]|##+\s*최종\s*답변|\*\*최종\s*답변\*\*)\s*[:：]?\s*', raw) if m_final: return m_final.end() low = raw.lower() if '전남광주 통합특별시 행정 AI를 OpenAI 호환 API로 사용하세요
JGOS API는 전남광주 통합특별시 행정 AI를 외부 서비스·앱에서 호출할 수 있는 공개 API입니다.
OpenAI chat/completions 규격과 100% 호환되어, 기존 OpenAI SDK에서 Base URL과 API 키만 바꾸면 즉시 사용할 수 있습니다.
| 항목 | 내용 |
|---|---|
| Base URL | https://jgos.vidraft.net/v1 |
| 모델 ID | jgos-31b |
| 핵심 기능 | 통합특별시 행정·민원 상담, 담당부서 연락처 자동 안내, 실시간 웹검색, 27개 시군구 지역정보 |
| 인증 | API 키 (Bearer 토큰) |
| 응답 | 일반(JSON) / 스트리밍(SSE) 모두 지원 |
모든 /v1 요청에는 발급받은 API 키를 Authorization 헤더에 담아야 합니다.
Authorization: Bearer jgos-live-여기에_발급받은_키또는 X-API-Key: jgos-live-... 헤더도 지원합니다.
POST /v1/chat/completions
curl https://jgos.vidraft.net/v1/chat/completions \\
-H "Authorization: Bearer jgos-live-여기에_발급받은_키" \\
-H "Content-Type: application/json" \\
-d '{
"model": "jgos-31b",
"messages": [
{"role": "user", "content": "전남광주 통합특별시 인구가 얼마나 되나요?"}
]
}'응답(JSON)의 choices[0].message.content에 답변이 담깁니다.
from openai import OpenAI
client = OpenAI(
api_key="jgos-live-여기에_발급받은_키",
base_url="https://jgos.vidraft.net/v1",
)
resp = client.chat.completions.create(
model="jgos-31b",
messages=[{"role": "user", "content": "여권 발급 절차 알려줘"}],
)
print(resp.choices[0].message.content)import requests
r = requests.post(
"https://jgos.vidraft.net/v1/chat/completions",
headers={"Authorization": "Bearer jgos-live-여기에_발급받은_키"},
json={"model": "jgos-31b",
"messages": [{"role": "user", "content": "광주 맛집 추천해줘"}]},
)
print(r.json()["choices"][0]["message"]["content"])const res = await fetch("https://jgos.vidraft.net/v1/chat/completions", {
method: "POST",
headers: {
"Authorization": "Bearer jgos-live-여기에_발급받은_키",
"Content-Type": "application/json",
},
body: JSON.stringify({
model: "jgos-31b",
messages: [{ role: "user", content: "순천만 국가정원 입장료 알려줘" }],
}),
});
const data = await res.json();
console.log(data.choices[0].message.content);| 필드 | 타입 | 기본 | 설명 |
|---|---|---|---|
| model | string | jgos-31b | 모델 ID (현재 jgos-31b) |
| messages | array | 필수 | OpenAI 형식 대화 배열 (role: system/user/assistant) |
| stream | bool | false | true 시 SSE 스트리밍 |
| temperature | float | 0.3 | 창의성 (0~1) |
| max_tokens | int | 1200 | 최대 응답 토큰 (상한 2048) |
| web | bool | false | JGOS 확장 · 실시간 웹검색 grounding |
| minwon | bool | false | JGOS 확장 · 담당부서·연락처 자동 안내(민원 집중) |
OpenAI SDK에서 web·minwon 같은 확장 파라미터는 extra_body={...}로 전달합니다.
minwon: true를 켜면, 질문을 분석해 광주 5구·전남 22시군 + 양 본청(총 29개 기관, 3.8만 행 조직도)에서 담당 부서와 직통 전화번호를 찾아 안내합니다. (전화번호는 실제 수집 데이터만 사용, 지어내지 않음)
resp = client.chat.completions.create(
model="jgos-31b",
messages=[{"role": "user", "content": "북구에서 쓰레기 무단투기 신고하려면 어디로 전화해요?"}],
extra_body={"minwon": True},
)
print(resp.choices[0].message.content)
# → 광주 북구 청소행정과 062-410-6509 등 실제 담당부서·연락처 안내stream = client.chat.completions.create(
model="jgos-31b",
messages=[{"role": "user", "content": "5·18 민주화운동에 대해 알려줘"}],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content or ""
print(delta, end="", flush=True)표준 OpenAI chat.completion.chunk 형식으로 전송되며 data: [DONE]으로 종료됩니다.
| 코드 | 의미 | 해결 |
|---|---|---|
| 400 | 잘못된 요청 | messages 필드·JSON 형식 확인 |
| 401 | API 키 없음/오류 | Authorization 헤더의 키 확인 |
| 502 | LLM 백엔드 오류 | 잠시 후 재시도 |
| 503 | 키 미설정 | 운영자 문의 |
🏛️ JGOS — 전남광주 통합특별시 행정 AI · powered by VIDRAFT
API 키 발급·제휴 문의: 통합특별시청 062-120 / VIDRAFT
데이터 없음
" all_keys: list = [] for r in records: for k in r: if k not in all_keys: all_keys.append(k) priority = STD_FIELDS_AX + ["출처파일"] keys = [k for k in priority if k in all_keys] + [k for k in all_keys if k not in priority] rows_html = "" for rec in records: status = rec.get("처리상태", "") cls = {"승인": "status-ok", "반려": "status-ng", "검토중": "status-review", "접수": "status-new"}.get(status, "") cells = "".join(f"파일 없음: {filename}
", "rows": 0}) ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" try: with open(fp, "rb") as f: data = f.read() if ext in ("xlsx", "xls"): recs = _parse_xlsx_ax(data, filename) elif ext == "csv": recs = _parse_csv_ax(data, filename) elif ext == "docx": recs = _parse_docx_ax(data, filename) elif ext == "pptx": recs = _parse_pptx_ax(data, filename) elif ext == "pdf": recs = _parse_pdf_ax(data, filename) elif ext in ("hwpx", "hwp"): recs = _parse_hwpx_ax(data, filename) else: recs = [] html = _records_to_html(recs, f"{filename} — {len(recs)}건") return JSONResponse({"ok": True, "filename": filename, "rows": len(recs), "html": html, "ext": ext}) except Exception as e: return JSONResponse({"ok": False, "html": f"오류: {e}
", "rows": 0}) @app.get("/api/demo/{filename}") async def demo_file(filename: str): fp = os.path.join(_DEMO_DIR, filename) if not os.path.isfile(fp): raise HTTPException(404, "demo file not found") mt = _mimetypes.guess_type(fp)[0] or "application/octet-stream" return FileResponse(fp, media_type=mt, filename=filename)