274 lines
9.8 KiB
Python
274 lines
9.8 KiB
Python
import tomllib
|
||
from typing import Dict, Any, List, Tuple
|
||
|
||
import aiohttp
|
||
import casbin
|
||
from casbin_tortoise_adapter import TortoiseAdapter
|
||
from nacl.signing import SigningKey
|
||
from sanic import Sanic, Request
|
||
from sanic.log import logger, Colors
|
||
from sanic.response import text
|
||
from socketio import AsyncClient
|
||
from tortoise.contrib.sanic import register_tortoise
|
||
|
||
import cmds
|
||
from gitea_model import WebHookIssueComment, WebHookIssue, GiteaEvent
|
||
from models import User, GiteaUser
|
||
from sio_model import Ctx, SioConfig, Message
|
||
from unit import sio_log_format, int2str, cas_log_fmt
|
||
|
||
app = Sanic('GiteaPush', ctx=Ctx)
|
||
|
||
|
||
def get_config() -> SioConfig:
|
||
with open('config.toml', 'rb') as f:
|
||
config = tomllib.load(f)
|
||
return SioConfig(**config)
|
||
|
||
|
||
SIO_CONFIG = get_config()
|
||
|
||
register_tortoise(
|
||
app, db_url=SIO_CONFIG.db_url, modules={"models": ["models", "casbin_tortoise_adapter"]}, generate_schemas=True
|
||
)
|
||
|
||
|
||
@app.before_server_start
|
||
async def setup_before_start(_app):
|
||
_app.ctx.sio_config = SIO_CONFIG
|
||
|
||
# 使用casbin策略管理
|
||
adapter = TortoiseAdapter()
|
||
e = casbin.AsyncEnforcer('./casbin_data/model.conf', adapter)
|
||
_app.ctx.e = e
|
||
|
||
t1 = await _app.ctx.e.add_policy('admin', '*')
|
||
t2 = await _app.ctx.e.add_policy('default', 'ping')
|
||
if t1 is True and t2 is True:
|
||
logger.info(cas_log_fmt('Init casbin rule success!'))
|
||
admins = int2str(_app.ctx.sio_config.admin)
|
||
for qid in admins:
|
||
if await _app.ctx.e.add_role_for_user(qid, 'admin'):
|
||
logger.debug(cas_log_fmt(f'Added {Colors.PURPLE}{qid}{Colors.YELLOW} to admin group'))
|
||
users = await _app.ctx.e.get_users_for_role('admin')
|
||
rm_user = set(users) ^ set(admins)
|
||
for u in list(rm_user):
|
||
if await _app.ctx.e.delete_user(u):
|
||
logger.debug(f'Delete {Colors.PURPLE}{u}{Colors.YELLOW} for group admin')
|
||
await _app.ctx.e.save_policy()
|
||
|
||
# 初始化sio
|
||
_app.ctx.sio = AsyncClient()
|
||
start_sio_listener()
|
||
|
||
await _app.ctx.sio.connect(_app.ctx.sio_config.host)
|
||
|
||
|
||
async def push_webhook2users(_app: Sanic, full_name: str, msg: str):
|
||
async with aiohttp.ClientSession() as session:
|
||
url = f'{app.ctx.sio_config.gitea_host}/api/v1/repos/{full_name}/subscribers?page=&limit='
|
||
async with session.get(url) as rps:
|
||
if rps.status == 200:
|
||
json = await rps.json()
|
||
q_user = set()
|
||
for g_user in json:
|
||
g_user = await GiteaUser.filter(id=g_user['id']).get_or_none()
|
||
if g_user:
|
||
_q_user = await g_user.qid.get()
|
||
q_user.add(_q_user.id)
|
||
for i in q_user:
|
||
message = Message(content=msg, room_id=i)
|
||
await app.ctx.sio.emit('sendMessage', message.to_json())
|
||
else:
|
||
logger.warn(rps)
|
||
|
||
|
||
@app.post('/receive')
|
||
async def receive(rqt: Request):
|
||
match (rqt.headers['X-Gitea-Event']):
|
||
case GiteaEvent.issues.value:
|
||
data = WebHookIssue(**rqt.json)
|
||
rsp_sender = f"{data.sender.username}"
|
||
rsp_title = f"[{data.repository.full_name}][{data.action}] {data.issue.title} (Issue #{data.issue.id})"
|
||
rsp_ctx = f"{data.issue.body}"
|
||
|
||
rsp_link = f"View it: {data.issue.html_url}"
|
||
cancel_subscribe = f"unsubscribe: {data.repository.html_url}"
|
||
|
||
rsp = f"{rsp_sender}\n{rsp_title}\n------\n{rsp_ctx}\n------\n{rsp_link}\n{cancel_subscribe}"
|
||
|
||
case GiteaEvent.issue_comment.value:
|
||
data = WebHookIssueComment(**rqt.json)
|
||
rsp_sender = f"{data.sender.username}"
|
||
rsp_title = f"Re:[{data.repository.full_name}][{data.action}] {data.issue.title} (Issue #{data.issue.id})"
|
||
rsp_ctx = f"{data.comment.body}"
|
||
|
||
rsp_link = f"View it: {data.comment.html_url}"
|
||
cancel_subscribe = f"unsubscribe: {data.repository.html_url}"
|
||
|
||
rsp = f"{rsp_sender}\n{rsp_title}\n------\n{rsp_ctx}\n------\n{rsp_link}\n{cancel_subscribe}"
|
||
|
||
case _:
|
||
data = None
|
||
rsp = "Unknown webhook type! Please contact the administrator."
|
||
logger.warn(rsp)
|
||
|
||
if data:
|
||
rqt.app.add_task(push_webhook2users(app, data.repository.full_name, rsp))
|
||
|
||
return text(rsp)
|
||
|
||
|
||
async def get_gitea_token(_app, form: dict, q_user: User):
|
||
async with aiohttp.ClientSession() as session:
|
||
url = f'{app.ctx.sio_config.gitea_host}/login/oauth/access_token'
|
||
async with session.post(url, data=form) as rps:
|
||
if rps.status == 200:
|
||
json = await rps.json()
|
||
act = json['access_token']
|
||
rft = json['refresh_token']
|
||
else:
|
||
logger.warn(f'Gitea: {rps}')
|
||
return f'绑定失败：{await rps.text()}'
|
||
url = f'{app.ctx.sio_config.gitea_host}/api/v1/user'
|
||
headers = {'Authorization': f'token {act}'}
|
||
async with session.get(url, headers=headers) as rps:
|
||
if rps.status == 200:
|
||
json = await rps.json()
|
||
gid = json['id']
|
||
login = json['login']
|
||
else:
|
||
logger.warn(f'Gitea: {rps}')
|
||
return f'绑定失败：{await rps.text()}'
|
||
|
||
g_user = await GiteaUser.filter(id=gid).get_or_none()
|
||
if g_user:
|
||
g_user.id = gid
|
||
g_user.name = login
|
||
g_user.qid = q_user
|
||
g_user.ac_tk = act
|
||
g_user.rfs_tk = rft
|
||
await g_user.save()
|
||
else:
|
||
await GiteaUser(id=gid, name=login, qid=q_user, ac_tk=act, rfs_tk=rft).save()
|
||
return f'绑定成功：QQ:{q_user.name}, Gitea:{login}'
|
||
|
||
|
||
@app.get('/redirect')
|
||
async def redirect(rqt: Request):
|
||
rqt_state = rqt.args.get('state')
|
||
code = rqt.args.get('code')
|
||
if rqt_state is None or code is None:
|
||
logger.warn(f'Gitea返回消息格式错误：{rqt.args}')
|
||
return text(f'返回信息错误：{rqt.args}')
|
||
|
||
q_user = await User.filter(state=rqt_state).get_or_none()
|
||
if q_user is None:
|
||
logger.warn(f'该请求不是从QQ中请求的链接')
|
||
return text('请先从QQ处申请绑定')
|
||
|
||
msg = Message(content='已在绑定中，可能因为意外而失败\n查询结果使用\n/gitea -ust', room_id=q_user.id)
|
||
await app.ctx.sio.emit('sendMessage', msg.to_json())
|
||
|
||
rqt.app.add_task(get_gitea_token(app, {'client_id': app.ctx.sio_config.client_id,
|
||
'client_secret': app.ctx.sio_config.client_secret,
|
||
'code': code,
|
||
'grant_type': 'authorization_code',
|
||
'redirect_uri': f'{app.ctx.sio_config.localhost}/redirect'},
|
||
q_user), name=str(q_user.id))
|
||
|
||
return text('success')
|
||
|
||
|
||
"""
|
||
以下为QQ监听部分
|
||
"""
|
||
|
||
|
||
def start_sio_listener():
|
||
@app.ctx.sio.on('connect')
|
||
def connect():
|
||
logger.info(f'{Colors.GREEN}icalingua 已连接{Colors.END}')
|
||
|
||
@app.ctx.sio.on('requireAuth')
|
||
async def require_auth(salt: str, versions: Dict[str, str]):
|
||
logger.info(
|
||
f"{Colors.BLUE}versions: {Colors.PURPLE}{versions} {Colors.BLUE}with type {type(salt)}|{salt=}{Colors.END}")
|
||
# 准备数据
|
||
sign = SigningKey(bytes.fromhex(app.ctx.sio_config.key))
|
||
signature = sign.sign(bytes.fromhex(salt))
|
||
|
||
# 发送数据
|
||
logger.info(f"{len(signature.signature)=} {type(signature.signature)=}{Colors.END}")
|
||
await app.ctx.sio.emit('auth', signature.signature)
|
||
logger.info(f"{Colors.BLUE}send auth emit{Colors.END}")
|
||
|
||
@app.ctx.sio.on('auth')
|
||
async def auth(data: Dict[str, Any]):
|
||
logger.info(f"auth: {data}")
|
||
|
||
@app.ctx.sio.on('authFailed')
|
||
async def auth_failed():
|
||
logger.warn(f"authFailed")
|
||
await app.ctx.sio.disconnect()
|
||
|
||
@app.ctx.sio.on('authSucceed')
|
||
def auth_succeed():
|
||
logger.info(f"authSucceed")
|
||
|
||
@app.ctx.sio.on('connect_error')
|
||
def connect_error(*args, **kwargs):
|
||
logger.warn(f"连接错误 {args}, {kwargs}")
|
||
|
||
@app.ctx.sio.on('updateRoom')
|
||
def update_room(data: Dict[str, Any]):
|
||
logger.debug(sio_log_format('update_room:', data))
|
||
|
||
@app.ctx.sio.on('deleteMessage')
|
||
def delete_message(message_id: str):
|
||
logger.debug(sio_log_format('delete_message:', message_id))
|
||
|
||
@app.ctx.sio.on('setMessages')
|
||
def set_messages(data: Dict[str, Any]):
|
||
logger.debug(f"{sio_log_format('set_messages:', data)}"
|
||
f"{sio_log_format('message_len:', len(data['messages']))}")
|
||
|
||
@app.ctx.sio.on('setAllRooms')
|
||
def set_all_rooms(rooms: List[Dict[str, Any]]):
|
||
logger.debug(f"{sio_log_format('set_all_rooms:', rooms)}"
|
||
f"{sio_log_format('len:', len(rooms))}")
|
||
|
||
@app.ctx.sio.on('setAllChatGroups')
|
||
def set_all_chat_groups(groups: List[Dict[str, Any]]):
|
||
logger.debug(f"{sio_log_format('set_all_chat_groups:', groups)}"
|
||
f"{sio_log_format('len:', len(groups))}")
|
||
|
||
@app.ctx.sio.on('notify')
|
||
def notify(data: List[Tuple[str, Any]]):
|
||
logger.debug(sio_log_format('notify:', data))
|
||
|
||
@app.ctx.sio.on('closeLoading')
|
||
def close_loading(_):
|
||
logger.debug(sio_log_format('close_loading', ''))
|
||
|
||
@app.ctx.sio.on('onlineData')
|
||
def online_data(data: Dict[str, Any]):
|
||
logger.debug(sio_log_format('online_data:', data))
|
||
|
||
@app.ctx.sio.on('*')
|
||
def catch_all(event, data):
|
||
logger.debug(sio_log_format('catch_all:', f'{event}|{data}'))
|
||
|
||
@app.ctx.sio.on('addMessage')
|
||
async def add_message(data: Dict[str, Any]):
|
||
sio_decorator = cmds.cmds(app, data)
|
||
try:
|
||
await sio_decorator.route2cmd_and_run()
|
||
except IndexError:
|
||
# 处理非文本消息
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app.run(host='0.0.0.0', port=80, dev=True)
|