wqhar

THE_SERVER.py

@@ -0,0 +1,254 @@
+import http.server as qqq
+import mysql.connector as ee2
+import json
+import datetime
+import os
+from dotenv import dotenv_values
+from typing import Callable, Dict, Any
+
+from things import (
+    funny_trycatch_wrapper,
+    process_tuples,
+)  # no way its the funny trycatch wrapper
+
+dotenv_values1 = dotenv_values()
+# he really did that
+# from mysql.connector import errorcode
+comectomKeys = ["user", "password", "host", "database"]
+comectomtrk = {
+    "db_user": "user",
+    "db_password": "password",
+    "db_host": "host",
+    "db_name": "database",
+}
+# def getnewdict(d:dict, l:list) -> dict:
+#     r = dict()
+#     for k in d:
+#         if k in l:
+#             r[k] = d[k]
+#     return r
+getnewdict = lambda d, l: {k: d[k] for k in d if k in l}
+# def getnewdicttranslate(d:dict, tltb:dict) -> dict:
+#     r = dict()
+#     for k in d:
+#         if k in tltb:
+#             r[tltb[k]] = d[k]
+#     return r
+getnewdicttranslate = lambda d, tltb: {tltb[k]: d[k] for k in d if k in tltb}
+
+
+connectionSettings = {
+    **getnewdicttranslate(dotenv_values1, comectomtrk),
+    "raise_on_warnings": True,
+}
+
+
+quer = "insert into tbale (kewod, id) values (%s, %s);"  # do not support custom key Only if
+vals = (
+    r"faf",
+    r"24",
+)
+
+
+def get_that_connection(
+    connectionSettings: dict,
+) -> tuple[any, BaseException | None]:  # the giving up is real
+    try:
+        connectio = ee2.connect(**connectionSettings)
+    except ee2.Error as e:
+        return (None, e)
+    else:
+        return (connectio, None)
+
+
+class ServerLogic(qqq.BaseHTTPRequestHandler):
+    def close2(self):
+        try:
+            self.server.server_close()
+        except BaseException as e:
+            print(e)
+        finally:
+            print("gato")
+            raise SystemExit("truly the /close")
+
+    def close(self):
+        funny_trycatch_wrapper(
+            tries=[(self.server.server_close)],
+            finallies=[(print, ("gato")), ('raise SystemExit("truly the /close")')],
+        )
+
+    def header_setup(
+        self, code: int = 200, headers: dict = {"Content-type": "text/html"}
+    ):
+        self.send_response(code)
+        if headers is not None:
+            for key in headers:
+                self.send_header(key, headers[key])  # why!
+        self.end_headers()
+
+    def do_GET2(self):
+        if self.path == "/close":
+            return self.close()
+
+        self.header_setup()
+        try:
+            connectio = ee2.connect(**connectionSettings)
+        except ee2.Error as e:
+            print(e)
+        else:
+            with connectio.cursor(buffered=True, raw=True) as cur:
+                try:
+                    res = cur.execute("select * from tbale")
+                    w = cur.fetchall()
+                except Exception as e:
+                    print(e)
+                else:
+                    print(w)
+                    self.wfile.write(bytes(w).decode())
+        return
+
+    def do_GET(self):
+        if self.path == "/close":
+            return self.close()
+        elif self.path == "/endp":
+            self.header_setup(200, {"Content-type": "application/json"})
+            (connectio, eror) = get_that_connection(connectionSettings)
+            if eror:
+                print("coneti eror: ", eror)
+                return
+            with connectio.cursor(buffered=True) as cur:
+                try:
+                    res = cur.execute("select * from tbale")
+                    w = cur.fetchall()
+                except Exception as e:
+                    print(e)
+                    return  # return ealy no else
+                c = process_tuples(w)
+                print(w, c, sep="\n")
+                self.wfile.write(str(c).encode())
+            return
+        else:
+            self.header_setup(404)
+            self.wfile.write(
+                b"""
+<h1>404</h1>
+<p>where u at</p>
+"""
+            )
+            return
+
+    def do_POST2(self):
+        if self.path == "/close":
+            return self.close()
+        self.header_setup()
+        content_length = int(self.headers["Content-Length"])
+        with self.rfile as r:
+            post_data_bytes = self.rfile.read(content_length)
+        self.wfile.write(
+            b"MY SERVER: The post data I received from the request has following data:\n"
+            + post_data_bytes
+        )
+        try:
+            data_dict = json.loads(post_data_bytes.decode("utf-8"))
+        except Exception as e:
+            print(e)
+        try:
+            connectio = ee2.connect(**connectionSettings)
+            connectio.autocommit = True  # EUYOEYOEYOEYOEYO
+        except ee2.Error as e:
+            print(e.errno, e.msg)
+        else:
+            with connectio.cursor(buffered=True) as cur:
+                try:
+                    res = cur.execute(
+                        "insert into tbale (kewod, id) values (%s, %s)",
+                        (
+                            data_dict["kewod"],
+                            data_dict["id"],
+                        ),
+                        multi=True,
+                    )
+
+                    print(res)
+                except Exception as e:
+                    print("execute err: ", e)
+                    self.wfile.write(b"\n is it very much Invalid." + str(e).encode())
+                else:
+                    print(cur)
+                    cur.execute("select * from tbale")
+                    self.wfile.write(
+                        b"\n We have this in the this: "
+                        + str.encode(str(cur.fetchall()))
+                    )
+        return
+
+    def do_POST(self):
+        match self.path:
+            case "/close":
+                return self.close()
+
+            case "/endp":
+                self.header_setup()
+                content_length = int(self.headers["Content-Length"])
+                with self.rfile as r:
+                    post_data_bytes = self.rfile.read(content_length)
+                self.wfile.write(
+                    b"MY SERVER: The post data I received from the request has following data:\n"
+                    + post_data_bytes
+                )
+                try:
+                    data_dict: dict = json.loads(post_data_bytes.decode("utf-8"))
+                except Exception as e:
+                    print(e)
+                (connectio, eror) = get_that_connection(connectionSettings)
+                if eror:
+                    print("coneti eror: ", eror)
+                    self.wfile.write(
+                        b"the thing cant even start like what f: " + str(eror).encode()
+                    )
+                    return
+                print((*data_dict.keys(), *data_dict.values()))
+                with connectio.cursor(prepared=True) as cur:
+                    try:
+                        res = cur.execute(
+                            "insert into tbale (kewod, expi) values (?, ?);",
+                            (data_dict["kewod"],data_dict["expi"],),
+                            multi=True,
+                        )
+
+                        print(res)
+                    except Exception as e:
+                        print("execute err: ", e)
+                        self.wfile.write(
+                            b"\n is it very much Invalid: " + str(e).encode()
+                        )
+                    else:
+                        print(cur)
+                        cur.execute("select * from tbale")
+                        self.wfile.write(b"\n We have this in the this: ")
+                        self.do_GET()
+            case _:
+                self.header_setup(403)
+                self.wfile.write(
+                    b"""
+    <h1>403</h1>
+    <p>tf u doin</p>
+    """
+                )
+
+        return
+
+
+def runServer():
+    server = qqq.HTTPServer(("", 8088), ServerLogic)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        server.server_close()
+        raise SystemExit("\nctrl c")
+    except BaseException as e:
+        print(e)
+
+
+if __name__ == "__main__":
+    runServer()

things.py

@@ -0,0 +1,173 @@
+from typing import Callable, Dict, Any, List, Tuple, Union
+from itertools import zip_longest
+import datetime
+
+class FunnyStuff:
+    @staticmethod
+    def funny(
+        f: Callable,
+        #    m: Dict["args": list(), "kwargs": dict()]
+        *a,
+        **b
+    ):
+        # a = m.get("args", [])
+        # b = m.get("kwargs", {})
+        return f(*a, **b)
+
+    @staticmethod
+    def funny2(t: tuple[Callable, tuple[Any, ...], Dict[str, Any]] = None):
+        f, a, k = t
+        # print(f, a, k)
+        return FunnyStuff.funny(f, *a, **k)
+
+    @staticmethod
+    def funny3(t: tuple[Callable, tuple[Any, ...], Dict[str, Any]] = None):
+        f, a, k = t
+        # print(f, a, k)
+        return f(*a, **k)
+
+
+# FunnyStuff.funny(print, 3, 4, 5, 6, 7, sep="")
+
+# FunnyStuff.funny2((print, (2, 4, 7, 9), {"sep": "\n"}))
+
+# FunnyStuff.funny3((print, (2, 4, "hello world", 9), {"sep": "   "}))
+
+
+def unpack_with_defaults(t, defaults=(None, (), {})):
+    # for i in zip_longest(t, defaults):
+    # print(i, "i in what")
+    return tuple(
+        item if item is not None else default
+        for item, default in zip_longest(t, defaults)
+    )
+
+
+def function_splitter(
+    t: tuple[Union[Callable, str], tuple[Any, ...], Dict[str, Any]] = (
+        None,
+        (),
+        {},
+    ),
+    *e,
+    **ae
+):
+    """inp: (a, (b)?, {c}?), *oa?, **okwa?,    
+    exec wrapper if len args == 1 + args type str,  
+    works like your function caller, (function, (args), {kwargs})  
+    function(*oa, **args, **okwa, **kwargs) # funny order huhg
+
+    Execute a function or a string using exec if it's the only argument.
+
+    :param t: Tuple containing a function or string, args, and kwargs.
+    :param e: Additional positional arguments.
+    :param ae: Additional keyword arguments.
+    :return: Result of the function call or exec.
+    """
+    if isinstance(t, str):
+        return exec(t)  # theres just no way
+    if not isinstance(t, tuple):
+        t = (t,)
+    f, a, k = unpack_with_defaults(t)
+    if not isinstance(a, tuple):
+        a = (a,)  # i dont know why but this is wack as hell
+    return f(*e, *a, **ae, **k)
+
+
+def funny_trycatch_wrapper(
+    tries: list[tuple[Callable| str, tuple[Any, ...] | Any, Dict[str, Any]]],
+    excepts: dict[
+        BaseException,
+        list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]],
+    ] = {BaseException: [(lambda x: print(f"Error: {x}"))]},
+    elses: list[tuple[Callable| str, tuple[Any, ...] | Any, Dict[str, Any]]] = [],
+    finallies: list[tuple[Callable| str, tuple[Any, ...] | Any, Dict[str, Any]]] = [],
+):
+    """
+    Try, except, else, and finally wrapper for multiple functions or exec strings.
+
+    :param tries: List of functions or strings to try.
+    :param excepts: Dictionary mapping exceptions to lists of functions or strings.
+    :param elses: List of functions or strings to run if no exceptions occur.
+    :param finallies: List of functions or strings to run after the try/except/else blocks.
+    """
+    try:
+        for oneTry in tries:
+            function_splitter(oneTry)
+    except BaseException as e:
+        for exception_type, handlers in excepts.items():
+            if isinstance(e, exception_type):
+                for handler in handlers:
+                    function_splitter(handler, e)
+                break
+
+
+    else:
+        for oneThing in elses:
+            function_splitter(oneThing)
+    finally:
+        for oneThing in finallies:
+            function_splitter(oneThing)
+
+
+thisone = "this string"
+thatone = "that string"
+tries = [
+    (
+        print,
+        ("hello world"),
+    ),
+    (print, (3 + 5, 9, 2 * 5, 33 / 3), {"sep": "     "}),
+    # ("1/0"),
+    # ("raise Exception('real exeption')"),
+]
+excepts = {
+    ZeroDivisionError: [(print, ("who divided by 0"))],
+    Exception: [(lambda e: print(f"what tf is this error {e} like what"))],
+}
+finallies = [
+    'print("finally")',
+]
+elses = [(print, "things gone well")]
+# funny_trycatch_wrapper(tries=tries, excepts=excepts, elses=elses, finallies=finallies)
+
+
+def funny_singular_Simple_trycatch(Onetry: tuple):
+    this = None
+    er = None
+    try:
+        this = function_splitter(Onetry)
+    except BaseException as e:
+        er = e
+    finally:
+        return (this, er)
+
+
+
+def process_tuples(tuples: List[Tuple[Any, ...]], action: str = "timestamp") -> List[Tuple[Any, ...]]:
+    """
+    Parse through a list of tuples and perform an action on datetime objects.
+    
+    :param tuples: List of tuples to process.
+    :param action: Action to perform on datetime objects. Can be "string" or "timestamp".
+    :return: List of processed tuples. 
+
+    > from open ai
+    """
+    processed_tuples = []
+
+    for tup in tuples:
+        processed_tuple = []
+        for item in tup:
+            if isinstance(item, datetime.datetime):
+                if action == "string":
+                    processed_tuple.append(item.strftime("%Y-%m-%d %H:%M:%S"))
+                elif action == "timestamp":
+                    processed_tuple.append(item.timestamp())
+                else:
+                    raise ValueError("Unsupported action. Use 'string' or 'timestamp'.")
+            else:
+                processed_tuple.append(item)
+        processed_tuples.append(tuple(processed_tuple))
+    
+    return processed_tuples