61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
from typing import Tuple, Dict
|
|
|
|
from db import pgclass
|
|
from utils import shareclass
|
|
|
|
# 獲取單一文章
|
|
def solo_article_fetcher(key:int) -> Dict | None:
|
|
table = pgclass.SQLarticle
|
|
ftab = pgclass.SQLfile
|
|
resfn = {}
|
|
|
|
with shareclass.Shared.db_get_session() as session:
|
|
# query
|
|
res = session.query(table).filter(table.id == key, table.mark == "visible").first()
|
|
if res is None: return None
|
|
|
|
# mapping
|
|
resfn.update({"id": res.id,
|
|
"ctx": res.ctx,
|
|
"igid": res.igid,
|
|
"mark": res.mark,
|
|
"reference": res.reference,
|
|
"hash": res.hash,
|
|
"created_at": res.created_at,
|
|
"ip": res.ip
|
|
})
|
|
|
|
# file
|
|
resfn["files"] = [ f[0] for f in session.query(ftab.id).filter(ftab.reference == res.hash).all() ]
|
|
|
|
return resfn
|
|
|
|
# 獲取檔案
|
|
def solo_file_fetcher(id:int) -> Dict | None:
|
|
table = pgclass.SQLarticle
|
|
ftab = pgclass.SQLfile
|
|
|
|
with shareclass.Shared.db_get_session() as session:
|
|
fres = session.query(ftab).filter(ftab.id == id).first()
|
|
if fres is None: # 檢查檔案是否存在
|
|
return None
|
|
|
|
article = session.query(table).filter(table.hash == fres.reference, table.mark == 'visible').first()
|
|
if article is None: # 檢查文章本體是否存在/可以閱覽
|
|
return None
|
|
|
|
# mapping
|
|
resfn = {
|
|
"type": fres.type,
|
|
"binary": fres.binary
|
|
}
|
|
|
|
return resfn
|
|
|
|
# 寫入IG狀態
|
|
def solo_article_updater(id:int, code:str):
|
|
table = pgclass.SQLarticle
|
|
with shareclass.Shared.db_get_session() as session:
|
|
res = session.query(table).filter(table.id == id).first()
|
|
res.igid = code
|
|
session.commit() |