TeamPGM/pyrogram
2
0
Fork 0
mirror of https://github.com/TeamPGM/pyrogram.git synced 2024-11-30 17:43:32 +00:00
Code Issues Packages Projects Releases Wiki Activity
fe2a8e6663
pyrogram/compiler/api/compiler.py
Dan a8d10443ea Pingu!
2018-01-05 02:38:29 +01:00

448 lines
17 KiB
Python
Raw Blame History

# Pyrogram - Telegram MTProto API Client Library for Python
# Copyright (C) 2017-2018 Dan Tès <https://github.com/delivrance>
#
# This file is part of Pyrogram.
#
# Pyrogram is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Pyrogram is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Pyrogram. If not, see <http://www.gnu.org/licenses/>.
import os
import re
import shutil
HOME = "compiler/api"
DESTINATION = "pyrogram/api"
notice_path = "NOTICE"
SECTION_RE = re.compile(r"---(\w+)---")
LAYER_RE = re.compile(r"//\sLAYER\s(\d+)")
COMBINATOR_RE = re.compile(r"^([\w.]+)#([0-9a-f]+)\s(?:.*)=\s([\w<>.]+);$", re.MULTILINE)
ARGS_RE = re.compile("[^{](\w+):([\w?!.<>]+)")
FLAGS_RE = re.compile(r"flags\.(\d+)\?")
FLAGS_RE_2 = re.compile(r"flags\.(\d+)\?([\w<>.]+)")
FLAGS_RE_3 = re.compile(r"flags:#")
core_types = ["int", "long", "int128", "int256", "double", "bytes", "string", "Bool"]
class Combinator:
def __init__(self, section: str, namespace: str, name: str, id: str, args: list, has_flags: bool, return_type: str):
self.section = section
self.namespace = namespace
self.name = name
self.id = id
self.args = args
self.has_flags = has_flags
self.return_type = return_type
def snek(s: str):
# https://stackoverflow.com/questions/1175208/elegant-python-function-to-convert-camelcase-to-snake-case
s = re.sub(r"(.)([A-Z][a-z]+)", r"\1_\2", s)
return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", s).lower()
def capit(s: str):
return "".join([i[0].upper() + i[1:] for i in s.split("_")])
def sort_args(args):
"""Put flags at the end"""
args = args.copy()
flags = [i for i in args if FLAGS_RE.match(i[1])]
for i in flags:
args.remove(i)
return args + flags
def start():
shutil.rmtree("{}/types".format(DESTINATION), ignore_errors=True)
shutil.rmtree("{}/functions".format(DESTINATION), ignore_errors=True)
with open("{}/source/auth_key.tl".format(HOME)) as auth, \
open("{}/source/sys_msgs.tl".format(HOME)) as system, \
open("{}/source/main_api.tl".format(HOME)) as api:
schema = (auth.read() + system.read() + api.read()).splitlines()
with open("{}/template/class.txt".format(HOME)) as f:
template = f.read()
with open(notice_path) as f:
notice = []
for line in f.readlines():
notice.append("# {}".format(line).strip())
notice = "\n".join(notice)
section = None
layer = None
namespaces = {"types": set(), "functions": set()}
combinators = []
for line in schema:
# Check for section changer lines
s = SECTION_RE.match(line)
if s:
section = s.group(1)
continue
# Save the layer version
l = LAYER_RE.match(line)
if l:
layer = l.group(1)
continue
combinator = COMBINATOR_RE.match(line)
if combinator:
name, id, return_type = combinator.groups()
namespace, name = name.split(".") if "." in name else ("", name)
args = ARGS_RE.findall(line)
# Pingu!
has_flags = not not FLAGS_RE_3.findall(line)
# Fix file and folder name collision
if name == "updates":
name = "update"
# Fix arg name being "self" (reserved keyword)
for i, item in enumerate(args):
if item[0] == "self":
args[i] = ("is_self", item[1])
if namespace:
namespaces[section].add(namespace)
combinators.append(
Combinator(
section,
namespace,
name,
"0x{}".format(id.zfill(8)),
args,
has_flags,
return_type
)
)
by_types = {}
for c in combinators:
return_type = capit(c.return_type)
if c.section == "types":
if return_type not in by_types:
by_types[return_type] = []
by_types[return_type].append(".".join(filter(None, [c.namespace, capit(c.name)])))
total = len(combinators)
current = 0
for c in combinators: # type: Combinator
print("Compiling APIs... [{}%] {}".format(
str(round(current * 100 / total)).rjust(3),
".".join(filter(None, [c.section, c.namespace, c.name]))
), end=" \r", flush=True)
current += 1
path = "{}/{}/{}".format(DESTINATION, c.section, c.namespace)
os.makedirs(path, exist_ok=True)
init = "{}/__init__.py".format(path)
if not os.path.exists(init):
with open(init, "w") as f:
f.write(notice + "\n\n")
with open(init, "a") as f:
f.write("from .{} import {}\n".format(snek(c.name), capit(c.name)))
sorted_args = sort_args(c.args)
arguments = ", " + ", ".join(
["{}{}".format(
i[0],
"=None" if FLAGS_RE.match(i[1]) else ""
) for i in sorted_args]
) if c.args else ""
fields = "\n ".join(
["self.{0} = {0} # {1}".format(i[0], i[1]) for i in c.args]
) if c.args else "pass"
docstring_args = []
for i, arg in enumerate(sorted_args):
arg_name, arg_type = arg
is_optional = arg_type.startswith("flags.")
arg_type = arg_type.split("?")[-1]
if arg_type in core_types:
if "int" in arg_type or arg_type == "long":
arg_type = ":obj:`int`"
elif arg_type == "double":
arg_type = ":obj:`float`"
else:
arg_type = ":obj:`{}`".format(arg_type.lower())
elif arg_type == "true":
arg_type = ":obj:`bool`"
else:
if arg_type.startswith("Vector"):
sub_type = arg_type.split("<")[1][:-1]
if sub_type in core_types:
if "int" in sub_type or sub_type == "long":
arg_type = "List of :obj:`int`"
elif sub_type == "double":
arg_type = "List of :obj:`float`"
else:
arg_type = "List of :obj:`{}`".format(sub_type.lower())
else:
arg_type = "List of :class:`pyrogram.api.types.{}`".format(
".".join(
sub_type.split(".")[:-1]
+ [capit(sub_type.split(".")[-1])]
)
)
else:
arg_type = ":class:`pyrogram.api.types.{}`".format(
".".join(
arg_type.split(".")[:-1]
+ [capit(arg_type.split(".")[-1])]
)
)
docstring_args.append(
"{}: {}{}".format(
arg_name,
arg_type,
" (optional)" if is_optional else ""
)
)
if docstring_args:
docstring_args = "Args:\n " + "\n ".join(docstring_args)
else:
docstring_args = "No parameters required."
docstring_args = "Attributes:\n ID (:obj:`int`): ``{}``\n\n ".format(c.id) + docstring_args
if c.section == "functions":
docstring_args += "\n\n Returns:\n "
if c.return_type in core_types:
if "int" in c.return_type or c.return_type == "long":
return_type = ":obj:`int`"
elif c.return_type == "double":
return_type = ":obj:`float`"
else:
return_type = ":obj:`{}`".format(c.return_type.lower())
else:
if c.return_type.startswith("Vector"):
sub_type = c.return_type.split("<")[1][:-1]
if sub_type in core_types:
if "int" in sub_type or sub_type == "long":
return_type = "List of :obj:`int`"
elif sub_type == "double":
return_type = "List of :obj:`float`"
else:
return_type = "List of :obj:`{}`".format(c.return_type.lower())
else:
if c.section == "functions":
try:
constructors = by_types[capit(sub_type)]
except KeyError:
return_type = "List of :class:`pyrogram.api.types.{}`".format(
".".join(
sub_type.split(".")[:-1]
+ [capit(sub_type.split(".")[-1])]
)
)
else:
constructors = ["List of :class:`pyrogram.api.types.{}`".format(
".".join(
i.split(".")[:-1]
+ [capit(i.split(".")[-1])]
)
) for i in constructors]
return_type = " | ".join(constructors)
else:
return_type = "List of :class:`pyrogram.api.types.{}`".format(
".".join(
sub_type.split(".")[:-1]
+ [capit(sub_type.split(".")[-1])]
)
)
else:
if c.section == "functions":
try:
constructors = by_types[capit(c.return_type)]
except KeyError:
return_type = ":class:`pyrogram.api.types.{}`".format(
".".join(filter(None, [c.namespace, capit(c.name)]))
)
else:
constructors = [":class:`pyrogram.api.types.{}`".format(
".".join(
i.split(".")[:-1]
+ [capit(i.split(".")[-1])]
)
) for i in constructors]
return_type = " | ".join(constructors)
else:
return_type = ":class:`pyrogram.api.types.{}`".format(
".".join(filter(None, [c.namespace, capit(c.name)]))
)
docstring_args += return_type
if c.section == "functions":
docstring_args += "\n\n Raises:\n :class:`pyrogram.Error`"
if c.has_flags:
write_flags = []
for i in c.args:
flag = FLAGS_RE.match(i[1])
if flag:
write_flags.append("flags |= (1 << {}) if self.{} is not None else 0".format(flag.group(1), i[0]))
write_flags = "\n ".join([
"flags = 0",
"\n ".join(write_flags),
"b.write(Int(flags))"
])
else:
write_flags = "# No flags"
read_flags = "flags = Int.read(b)" if c.has_flags else "# No flags"
write_types = read_types = ""
for arg_name, arg_type in c.args:
flag = FLAGS_RE_2.findall(arg_type)
if flag:
index, flag_type = flag[0]
if flag_type == "true":
read_types += "\n "
read_types += "{} = True if flags & (1 << {}) else False".format(arg_name, index)
elif flag_type in core_types:
write_types += "\n "
write_types += "if self.{} is not None:\n ".format(arg_name)
write_types += "b.write({}(self.{}))\n ".format(flag_type.title(), arg_name)
read_types += "\n "
read_types += "{} = {}.read(b) if flags & (1 << {}) else None".format(
arg_name, flag_type.title(), index
)
elif "vector" in flag_type.lower():
sub_type = arg_type.split("<")[1][:-1]
write_types += "\n "
write_types += "if self.{} is not None:\n ".format(arg_name)
write_types += "b.write(Vector(self.{}{}))\n ".format(
arg_name, ", {}".format(sub_type.title()) if sub_type in core_types else ""
)
read_types += "\n "
read_types += "{} = Object.read(b{}) if flags & (1 << {}) else []\n ".format(
arg_name, ", {}".format(sub_type.title()) if sub_type in core_types else "", index
)
else:
write_types += "\n "
write_types += "if self.{} is not None:\n ".format(arg_name)
write_types += "b.write(self.{}.write())\n ".format(arg_name)
read_types += "\n "
read_types += "{} = Object.read(b) if flags & (1 << {}) else None\n ".format(
arg_name, index
)
else:
if arg_type in core_types:
write_types += "\n "
write_types += "b.write({}(self.{}))\n ".format(arg_type.title(), arg_name)
read_types += "\n "
read_types += "{} = {}.read(b)\n ".format(arg_name, arg_type.title())
elif "vector" in arg_type.lower():
sub_type = arg_type.split("<")[1][:-1]
write_types += "\n "
write_types += "b.write(Vector(self.{}{}))\n ".format(
arg_name, ", {}".format(sub_type.title()) if sub_type in core_types else ""
)
read_types += "\n "
read_types += "{} = Object.read(b{})\n ".format(
arg_name, ", {}".format(sub_type.title()) if sub_type in core_types else ""
)
else:
write_types += "\n "
write_types += "b.write(self.{}.write())\n ".format(arg_name)
read_types += "\n "
read_types += "{} = Object.read(b)\n ".format(arg_name)
with open("{}/{}.py".format(path, snek(c.name)), "w") as f:
f.write(
template.format(
notice=notice,
class_name=capit(c.name),
docstring_args=docstring_args,
object_id=c.id,
arguments=arguments,
fields=fields,
read_flags=read_flags,
read_types=read_types,
write_flags=write_flags,
write_types=write_types,
return_arguments=", ".join([i[0] for i in sorted_args])
)
)
with open("{}/all.py".format(DESTINATION), "w") as f:
f.write(notice + "\n\n")
f.write("layer = {}\n\n".format(layer))
f.write("objects = {")
for c in combinators:
path = ".".join(filter(None, [c.section, c.namespace, capit(c.name)]))
f.write("\n {}: \"{}\",".format(c.id, path))
f.write("\n 0xbc799737: \"core.BoolFalse\",")
f.write("\n 0x997275b5: \"core.BoolTrue\",")
f.write("\n 0x56730bcc: \"core.Null\",")
f.write("\n 0x1cb5c415: \"core.Vector\",")
f.write("\n 0x73f1f8dc: \"core.MsgContainer\",")
f.write("\n 0xae500895: \"core.FutureSalts\",")
f.write("\n 0x0949d9dc: \"core.FutureSalt\",")
f.write("\n 0x3072cfa1: \"core.GzipPacked\",")
f.write("\n 0x5bb8e511: \"core.Message\"")
f.write("\n}\n")
for k, v in namespaces.items():
with open("{}/{}/__init__.py".format(DESTINATION, k), "a") as f:
f.write("from . import {}\n".format(", ".join([i for i in v])) if v else "")
if "__main__" == __name__:
HOME = "."
DESTINATION = "../../pyrogram/api"
notice_path = "../../NOTICE"
start()
Reference in New Issue View Git Blame Copy Permalink