FGO/artemis
3
2
Fork 0
forked from Dniel97/artemis
Code Pull Requests Activity

remove unused imports

Browse Source
This commit is contained in:
Hay1tsme 2024-01-09 15:59:58 -05:00
parent f8c77e69ed
commit 2d95e29f3c
8 changed files with 1 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core/aimedb.py
View File

@ -1,7 +1,5 @@
from twisted.internet.protocol import Factory, Protocol
import logging, coloredlogs
from Crypto.Cipher import AES
import struct
from typing import Dict, Tuple, Callable, Union, Optional
import asyncio
from logging.handlers import TimedRotatingFileHandler

3
core/mucha.py
View File

@ -1,7 +1,6 @@
from typing import Dict, Any, Optional, List
from typing import Dict, Any, Optional
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
from twisted.web import resource
from starlette.requests import Request
from datetime import datetime
from Crypto.Cipher import Blowfish

1
titles/idac/echo.py
View File

@ -1,7 +1,6 @@
import logging
import socket
from twisted.internet.protocol import DatagramProtocol
from socketserver import BaseRequestHandler, TCPServer
from typing import Tuple

2
titles/idac/index.py
View File

@ -19,8 +19,6 @@ from titles.idac.season2 import IDACSeason2
from titles.idac.config import IDACConfig
from titles.idac.const import IDACConstants
from titles.idac.echo import IDACEchoUDP
from titles.idac.matching import IDACMatching
class IDACServlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:

72
titles/idac/matching.py
View File

@ -1,72 +0,0 @@
import json
import logging
from typing import Dict
from twisted.web import resource
from core import CoreConfig
from titles.idac.season2 import IDACBase
from titles.idac.config import IDACConfig
class IDACMatching(resource.Resource):
isLeaf = True
def __init__(self, cfg: CoreConfig, game_cfg: IDACConfig) -> None:
self.core_config = cfg
self.game_config = game_cfg
self.base = IDACBase(cfg, game_cfg)
self.logger = logging.getLogger("idac")
self.queue = 0
def get_matching_state(self):
if self.queue >= 1:
self.queue -= 1
return 0
else:
return 1
def render_POST(self, req) -> bytes:
url = req.uri.decode()
req_data = json.loads(req.content.getvalue().decode())
header_application = self.decode_header(req.getAllHeaders())
user_id = int(header_application["session"])
# self.getMatchingStatus(user_id)
self.logger.info(
f"IDAC Matching request from {req.getClientIP()}: {url} - {req_data}"
)
resp = {"status_code": "0"}
if url == "/regist":
self.queue = self.queue + 1
elif url == "/status":
if req_data.get("cancel_flag"):
self.queue = self.queue - 1
self.logger.info(
f"IDAC Matching endpoint {req.getClientIP()} had quited"
)
resp = {
"status_code": "0",
# Only IPv4 is supported
"host": self.game_config.server.matching_host,
"port": self.game_config.server.matching_p2p,
"room_name": "INDTA",
"state": 1,
}
self.logger.debug(f"Response {resp}")
return json.dumps(resp, ensure_ascii=False).encode("utf-8")
def decode_header(self, data: Dict) -> Dict:
app: str = data[b"application"].decode()
ret = {}
for x in app.split(", "):
y = x.split("=")
ret[y[0]] = y[1].replace('"', "")
return ret

4
titles/idz/echo.py
View File

@ -1,9 +1,5 @@
from twisted.internet.protocol import DatagramProtocol
import logging
from core.config import CoreConfig
from .config import IDZConfig
class IDZEcho:
def connection_made(self, transport):
self.transport = transport

2
titles/ongeki/frontend.py
View File

@ -5,8 +5,6 @@ import jinja2
from starlette.requests import Request
from starlette.responses import Response, RedirectResponse
from os import path
from twisted.web.util import redirectTo
from twisted.web.server import Session
from core.frontend import FE_Base, UserSession
from core.config import CoreConfig

66
titles/pokken/services.py
View File

@ -1,66 +0,0 @@
from twisted.internet.interfaces import IAddress
from twisted.internet.protocol import Protocol
from autobahn.twisted.websocket import WebSocketServerProtocol, WebSocketServerFactory
from autobahn.websocket.types import ConnectionRequest
from typing import Dict
import logging
import json
from core.config import CoreConfig
from .config import PokkenConfig
from .base import PokkenBase
class PokkenAdmissionProtocol(WebSocketServerProtocol):
def __init__(self, cfg: CoreConfig, game_cfg: PokkenConfig):
super().__init__()
self.core_config = cfg
self.game_config = game_cfg
self.logger = logging.getLogger("pokken")
self.base = PokkenBase(cfg, game_cfg)
def onConnect(self, request: ConnectionRequest) -> None:
self.logger.debug(f"Admission: Connection from {request.peer}")
def onClose(self, wasClean: bool, code: int, reason: str) -> None:
self.logger.debug(f"Admission: Connection with {self.transport.getPeer().host} closed {'cleanly ' if wasClean else ''}with code {code} - {reason}")
def onMessage(self, payload, isBinary: bool) -> None:
msg: Dict = json.loads(payload)
self.logger.debug(f"Admission: Message from {self.transport.getPeer().host}:{self.transport.getPeer().port} - {msg}")
api = msg.get("api", "noop")
handler = getattr(self.base, f"handle_admission_{api.lower()}")
resp = handler(msg, self.transport.getPeer().host)
if resp is None:
resp = {}
if "type" not in resp:
resp['type'] = "res"
if "data" not in resp:
resp['data'] = {}
if "api" not in resp:
resp['api'] = api
if "result" not in resp:
resp['result'] = 'true'
self.logger.debug(f"Websocket response: {resp}")
self.sendMessage(json.dumps(resp).encode(), isBinary)
class PokkenAdmissionFactory(WebSocketServerFactory):
protocol = PokkenAdmissionProtocol
def __init__(
self,
cfg: CoreConfig,
game_cfg: PokkenConfig
) -> None:
self.core_config = cfg
self.game_config = game_cfg
super().__init__(f"ws://{self.game_config.server.hostname}:{self.game_config.ports.admission}")
def buildProtocol(self, addr: IAddress) -> Protocol:
p = self.protocol(self.core_config, self.game_config)
p.factory = self
return p