386 lines
12 KiB
Python
386 lines
12 KiB
Python
from typing import Tuple, Dict, List
|
||
from datetime import datetime
|
||
import time
|
||
import secrets
|
||
import hashlib
|
||
import os
|
||
|
||
from flask import make_response, Response, abort, request
|
||
from sqlalchemy.orm import sessionmaker
|
||
from sqlalchemy import desc, func, update, Engine, text, delete
|
||
import pytz
|
||
|
||
from utils import pgclass, setting_loader, s3helper, logger
|
||
from utils.misc import error
|
||
from protobuf_files import niming_pb2
|
||
|
||
class DB:
|
||
_engine = None
|
||
|
||
@classmethod
|
||
def __init__(cls, engine):
|
||
cls._engine:Engine = engine
|
||
|
||
@classmethod
|
||
def getsession(cls):
|
||
Session = sessionmaker(bind=cls._engine)
|
||
return Session()
|
||
|
||
db:DB = None
|
||
|
||
|
||
# 上傳單一文章
|
||
def solo_article_uploader(content:str, file_list, fmimes:List[str]) -> Tuple[int, str]:
|
||
# loadset
|
||
opt = setting_loader.loadset()
|
||
chk_before_post = opt["Check_Before_Post"]
|
||
|
||
# hash
|
||
seed = content + str(time.time()) + str(secrets.token_urlsafe(nbytes=16))
|
||
hash = hashlib.sha256(seed.encode()).hexdigest()
|
||
|
||
# IP
|
||
ip = request.remote_addr
|
||
|
||
# ig posting (only article)
|
||
if chk_before_post:
|
||
igid = None
|
||
# Go posting
|
||
igid = None
|
||
# Coming Soon...
|
||
|
||
# mark
|
||
if chk_before_post: mark = "pending"
|
||
else: mark = "visible"
|
||
|
||
# posting
|
||
article = pgclass.SQLarticle
|
||
article_mark = pgclass.SQLmark
|
||
article_metadata = pgclass.SQLmeta
|
||
result_id = 0
|
||
try:
|
||
with db.getsession() as session:
|
||
# file processor
|
||
fnlist, err = s3helper.multi_file_uploader(file_list, fmimes)
|
||
if err:
|
||
return 0, ""
|
||
|
||
# meta processor
|
||
metaa = article_metadata(ip=ip,
|
||
igid=igid,
|
||
hash=hash)
|
||
session.add(metaa)
|
||
|
||
# article processor
|
||
posta = article(content=content,
|
||
hash=hash,
|
||
file_list=fnlist)
|
||
session.add(posta)
|
||
|
||
# mark processor
|
||
marka = article_mark(hash=hash,
|
||
mark=mark)
|
||
session.add(marka)
|
||
|
||
# commit
|
||
session.commit()
|
||
result_id = int(posta.id)
|
||
|
||
# logger
|
||
logger.logger("newpost", "New post (id=%d): %s"%(result_id, mark))
|
||
|
||
return result_id, hash
|
||
except Exception as e:
|
||
print(e)
|
||
return 0, ""
|
||
|
||
|
||
# 上傳單一留言
|
||
def solo_comment_uploader(content:str, ref:int) -> Tuple[int | str, str]:
|
||
# loadset
|
||
opt = setting_loader.loadset()
|
||
chk_before_post = opt["Check_Before_Post"]
|
||
|
||
# hash
|
||
seed = content + str(time.time()) + str(secrets.token_urlsafe(nbytes=16))
|
||
hash = hashlib.sha256(seed.encode()).hexdigest()
|
||
sha1 = hashlib.sha1(seed.encode()).hexdigest()
|
||
|
||
# IP
|
||
ip = request.remote_addr
|
||
|
||
# mark
|
||
if chk_before_post: mark = "pending"
|
||
else: mark = "visible"
|
||
|
||
# posting
|
||
article = pgclass.SQLarticle
|
||
article_mark = pgclass.SQLmark
|
||
try:
|
||
with db.getsession() as session:
|
||
# article processor
|
||
cda = {
|
||
"content":content,
|
||
"ip":ip,
|
||
"hash":hash,
|
||
"created_at":datetime.now(pytz.timezone(os.getenv("TIMEZONE"))),
|
||
"sha1":sha1
|
||
}
|
||
|
||
session.execute(
|
||
update(article)
|
||
.where(article.id == ref)
|
||
.values(comment_list=article.comment_list + [cda])
|
||
)
|
||
|
||
# mark processor
|
||
marka = article_mark(hash=hash,
|
||
mark=mark)
|
||
session.add(marka)
|
||
|
||
# commit
|
||
session.commit()
|
||
|
||
# logger
|
||
logger.logger("newcomment", "New comment %s points to %d: %s"%(sha1, ref, mark))
|
||
|
||
return sha1, hash
|
||
except Exception as e:
|
||
return 0, ""
|
||
|
||
|
||
# role (general) (owner) (admin)
|
||
# 獲取單一文章
|
||
def solo_article_fetcher(role:str, key) -> Tuple[Dict, int]: # admin, owner, general
|
||
with db.getsession() as session:
|
||
# query
|
||
stmt = "SELECT posts.id AS posts_id, \
|
||
posts.content AS posts_content, \
|
||
posts.file_list AS posts_file_list, \
|
||
article_meta.igid AS article_meta_igid, \
|
||
posts.comment_list AS posts_comment_list, \
|
||
posts.hash AS posts_hash, \
|
||
article_meta.ip AS article_meta_ip \
|
||
FROM posts \
|
||
JOIN mark ON mark.hash = posts.hash \
|
||
JOIN article_meta ON article_meta.hash = posts.hash "
|
||
|
||
if role == "owner":
|
||
stmt += "WHERE posts.id = :id AND posts.hash = :hash"
|
||
result = session.execute(text(stmt), {"id":key[1], "hash":key[0]})
|
||
elif role == "admin":
|
||
stmt += "WHERE posts.id = :id"
|
||
result = session.execute(text(stmt), {"id":key})
|
||
elif role == "general":
|
||
stmt += "WHERE posts.id = :id AND mark.mark = 'visible'"
|
||
result = session.execute(text(stmt), {"id":key})
|
||
res = result.first()
|
||
if res is None:
|
||
return abort(404)
|
||
|
||
# mapping
|
||
one = {
|
||
"id": res[0],
|
||
"content": res[1],
|
||
"files_hash": res[2],
|
||
"igid": res[3],
|
||
}
|
||
if res[4]:
|
||
one["comments_hash"] = [ c.sha1 for c in res[4] ]
|
||
|
||
if role == "admin":
|
||
one["ip"] = res[6]
|
||
if role == "owner" or role == "admin":
|
||
one["hash"] = res[5]
|
||
|
||
return one, 200
|
||
|
||
|
||
# role (general) (owner) (admin)
|
||
# 獲取單一留言
|
||
def solo_comment_fetcher(role:str, key) -> Tuple[Dict, int]: # admin, owner, general
|
||
with db.getsession() as session:
|
||
# query
|
||
stmt = "SELECT posts.id AS parent, c.* \
|
||
FROM posts \
|
||
JOIN mark ON mark.hash = posts.hash \
|
||
JOIN unnest(posts.comment_list) AS c ON 1=1 "
|
||
if role == "general":
|
||
# 對一般用戶，sha1查詢，確保本體可見
|
||
stmt += " WHERE c.sha1 = :key AND mark.mark = 'visible'"
|
||
arta = session.execute(text(stmt), {'key':key}).first()
|
||
elif role == "owner":
|
||
# 對發文者，sha256查詢
|
||
stmt += " WHERE c.hash = :key AND c.sha1 = :sha1"
|
||
arta = session.execute(text(stmt), {'key':key[0], 'sha1':key[1]}).first()
|
||
elif role == "admin":
|
||
# 對管理員，sha1查詢
|
||
stmt += " WHERE c.sha1 = :key"
|
||
arta = session.execute(text(stmt), {'key':key}).first()
|
||
if arta is None:
|
||
return abort(404)
|
||
|
||
# mapping
|
||
one = {
|
||
"content": arta[1],
|
||
"sha1": arta[5]
|
||
}
|
||
|
||
if role == "admin":
|
||
one["ip"] = arta[2]
|
||
if role == "owner" or role == "admin":
|
||
one["hash"] = arta[3]
|
||
|
||
return one, 200
|
||
|
||
|
||
# 獲取文章列表
|
||
def multi_article_fetcher(role:str, page:str, count:int) -> Tuple[bytes, int]: # general, admin
|
||
# checker
|
||
if page is None or not page.isdigit():
|
||
return abort(400)
|
||
page = int(page)*count
|
||
|
||
article = pgclass.SQLarticle
|
||
article_meta = pgclass.SQLmeta
|
||
article_mark = pgclass.SQLmark
|
||
resfn = niming_pb2.FetchResponse()
|
||
|
||
with db.getsession() as session:
|
||
# query
|
||
res = session.query(article.id, article.content, article.file_list, article_meta.igid, article.hash, article_meta.ip)
|
||
res = res.join(article_meta, article_meta.hash==article.hash)
|
||
res = res.join(article_mark, article_mark.hash==article.hash)
|
||
if role == "general":
|
||
res = res.filter(article_mark.mark == "visible")
|
||
res = res.order_by(desc(article.id)).offset(page).limit(count).all()
|
||
|
||
# mapping
|
||
for r in res:
|
||
one = niming_pb2.FetchResponse.Message(
|
||
id = r[0],
|
||
content = r[1],
|
||
files_hash = r[2],
|
||
igid = r[3],
|
||
)
|
||
if role == "admin": # 如果是管理員 多給ip 跟 hash # proto那邊沒支援
|
||
one.hash = r[4]
|
||
one.ip = r[5]
|
||
resfn.posts.append(one)
|
||
|
||
return resfn.SerializeToString(), 200
|
||
|
||
|
||
# 刪除單一文章
|
||
def solo_article_remover(role:str, hash:str=None, id:int=None) -> Tuple[Dict, int]: # admin, owner
|
||
key = None
|
||
if role == "admin": key = id
|
||
elif role == "owner": key = (hash, id)
|
||
|
||
article = pgclass.SQLarticle
|
||
article_mark = pgclass.SQLmark
|
||
with db.getsession() as session:
|
||
# 獲取本體
|
||
pres = session.query(article.id, article.hash, article_mark.mark, article.file_list).join(article_mark, article_mark.hash==article.hash)
|
||
if role == "admin":
|
||
pres = pres.filter(article.id == key).first()
|
||
elif role == "owner":
|
||
pres = pres.filter(article.id == key[1], article.hash == key[0]).first()
|
||
if pres is None: # 如果本體不存在
|
||
return abort(404)
|
||
|
||
# 獲取本體的留言們(hash)
|
||
stmt="SELECT c.hash as chash " \
|
||
+ "FROM posts, unnest(posts.comment_list) AS c " \
|
||
+ "WHERE posts.id = :id"
|
||
cres = session.execute(text(stmt), {'id':pres[0]}).all()
|
||
|
||
# 刪除本體
|
||
stmt = delete(article).where(article.hash == pres[1])
|
||
session.execute(stmt)
|
||
|
||
# 刪除 mark (本體 & 留言)
|
||
stmt = delete(article_mark).where(article_mark.hash == pres[1])
|
||
session.execute(stmt)
|
||
for c in cres:
|
||
stmt = delete(article_mark).where(article_mark.hash == c[0])
|
||
session.execute(stmt)
|
||
|
||
# 刪除檔案
|
||
err = s3helper.multi_file_remover(pres[3])
|
||
if err:
|
||
return abort(500)
|
||
|
||
session.commit()
|
||
|
||
# logger
|
||
logger.logger("delpost", "Delete post (id=%d): last_status=%s"
|
||
%(int(pres[0]), str(pres[2])))
|
||
|
||
return {"id":pres[0], "mark":pres[2]}, 200
|
||
|
||
|
||
# 刪除單一留言
|
||
def solo_comment_remover(role:str, hash:str=None, sha1:str=None) -> Tuple[Dict, int]:
|
||
key = None
|
||
if role == "admin": key = sha1
|
||
elif role == "owner": key = (hash, sha1)
|
||
|
||
article_mark = pgclass.SQLmark
|
||
with db.getsession() as session:
|
||
# 獲取留言本體
|
||
stmt="SELECT posts.id AS parent, c.sha1, c.hash " \
|
||
+ "FROM posts, unnest(posts.comment_list) AS c "
|
||
if role == "admin":
|
||
stmt += "WHERE c.sha1 = :sha1"
|
||
cres = session.execute(text(stmt), {'sha1':key}).first()
|
||
elif role == 'owner':
|
||
stmt += "WHERE c.sha1 = :sha1 AND c.hash = :hash"
|
||
cres = session.execute(text(stmt), {'sha1':key[1], 'hash':key[0]}).first()
|
||
if cres is None: # 如果不存在
|
||
return abort(404)
|
||
|
||
# 刪除本體
|
||
stmt="UPDATE posts " \
|
||
+"SET comment_list = ARRAY(" \
|
||
+"SELECT c " \
|
||
+"FROM unnest(comment_list) AS c " \
|
||
+"WHERE (c.sha1, c.hash) != (:sha1, :hash)" \
|
||
+")"
|
||
session.execute(text(stmt), {'sha1':cres[1], 'hash':cres[2]})
|
||
|
||
# 刪除 mark (本體 & 留言)
|
||
mark = session.query(article_mark.mark).filter(article_mark.hash == cres[2])
|
||
stmt = delete(article_mark).where(article_mark.hash == cres[2])
|
||
session.execute(stmt)
|
||
|
||
session.commit()
|
||
|
||
logger.logger("delcomment", "Delete comment (sha1=%s): last_status=%s"
|
||
%(cres[1], str(mark)))
|
||
|
||
return {"sha1":cres[1], "mark":mark}, 200
|
||
|
||
|
||
# 獲取檔案
|
||
def solo_file_fetcher(role:str, fnhash:str) -> Tuple[Response, int]: # general, admin
|
||
with db.getsession() as session:
|
||
arta="SELECT posts.id, posts.hash, mark.mark, f FROM posts " \
|
||
+"JOIN unnest(file_list) AS f ON 1=1 " \
|
||
+"JOIN mark ON posts.hash = mark.hash " \
|
||
+"WHERE f = :fnhash "
|
||
if role == "general":
|
||
arta += "AND mark.mark = 'visible'"
|
||
arta = session.execute(text(arta), {'fnhash':fnhash}).first()
|
||
if arta is None: # 檢查文章本體是否存在/可以閱覽
|
||
return error("File not found"), 404
|
||
|
||
# fetch file
|
||
f, err = s3helper.solo_file_fetcher(fnhash)
|
||
if err:
|
||
return error("File not found"), 404
|
||
resp = make_response(f["binary"])
|
||
resp.headers.set("Content-Type", f["mime"])
|
||
resp.headers.set("Content-Disposition", f"attachment; filename=file_{fnhash}")
|
||
return resp, 200
|