minor changes: it takes a long time cuz im bad

do not know if the thing runs properly nah
This commit is contained in:
lda committed 2024-07-15 08:32:31 +07:00
1 parent 853dd7871d
commit 81ead9bb32
4 files changed
+230 -113

No files matched your search

+156 -95
View File
@@ -1,3 +1,9 @@
"""
go dam
???????? code
"""
import http.server as qqq import http.server as qqq
import mysql.connector as ee2 import mysql.connector as ee2
import json import json
@@ -11,6 +17,7 @@ from dotenv import dotenv_values
from things import ( from things import (
funny_trycatch_wrapper, funny_trycatch_wrapper,
process_tuples, process_tuples,
funny_tuples_to_dicts_processor,
) # no way its the funny trycatch wrapper ) # no way its the funny trycatch wrapper
dotenv_values1 = dotenv_values() dotenv_values1 = dotenv_values()
@@ -23,6 +30,7 @@ comectomtrk = {
"db_host": "host", "db_host": "host",
"db_name": "database", "db_name": "database",
} }
# def getnewdict(d:dict, l:list) -> dict: # def getnewdict(d:dict, l:list) -> dict:
# r = dict() # r = dict()
# for k in d: # for k in d:
@@ -30,6 +38,7 @@ comectomtrk = {
# r[k] = d[k] # r[k] = d[k]
# return r # return r
getnewdict = lambda d, l: {k: d[k] for k in d if k in l} getnewdict = lambda d, l: {k: d[k] for k in d if k in l}
# def getnewdicttranslate(d:dict, tltb:dict) -> dict: # def getnewdicttranslate(d:dict, tltb:dict) -> dict:
# r = dict() # r = dict()
# for k in d: # for k in d:
@@ -64,14 +73,78 @@ def get_that_connection(
class ServerLogic(qqq.BaseHTTPRequestHandler): class ServerLogic(qqq.BaseHTTPRequestHandler):
def __close2(self): # def __do_POST2(self):
try: # if self.path == "/close":
self.server.server_close() # return self.close()
except BaseException as e: # self.header_setup()
print(e) # content_length = int(self.headers["Content-Length"])
finally: # with self.rfile as r:
print("gato") # post_data_bytes = self.rfile.read(content_length)
raise SystemExit("truly the /close") # 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_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 __close2(self):
# try:
# self.server.server_close()
# except BaseException as e:
# print(e)
# finally:
# print("gato")
# raise SystemExit("truly the /close")
def youJustHaveToDoTheseStuff(self): def youJustHaveToDoTheseStuff(self):
print(self.headers, "hd") print(self.headers, "hd")
@@ -91,68 +164,63 @@ class ServerLogic(qqq.BaseHTTPRequestHandler):
self.send_header(key, headers[key]) # why! self.send_header(key, headers[key]) # why!
self.end_headers() 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): def do_GET(self):
self.youJustHaveToDoTheseStuff() self.youJustHaveToDoTheseStuff()
if self.path == "/close": if self.path == "/close":
return self.close() return self.close()
if self.path == "/": if self.path == "/":
self.header_setup() self.header_setup()
with open("index.html", "rb") as file: with open("index.html", "rb") as file:
self.wfile.write(file.read()) self.wfile.write(file.read())
return return
(connectio, eror) = get_that_connection(connectionSettings) (connectio, eror) = get_that_connection(connectionSettings)
if eror: if eror:
print("coneti eror: ", eror) print("coneti eror: ", eror)
return return
# HE. MUST. AVOID. THE. TRYCATCH
with connectio.cursor(buffered=True) as cur: with connectio.cursor(buffered=True) as cur:
try: try:
res = cur.execute("select * from tbale") res = cur.execute("select * from tbale")
w = cur.fetchall() w = cur.fetchall()
anotherone = cur.execute(
"select column_name from INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'tbale'"
)
anotherone2 = cur.fetchall()
except Exception as e: except Exception as e:
print(e) print(e)
return # return ealy no else return # return ealy no else
if self.path == "/endp": if self.path == "/endp":
self.header_setup(200, {"Content-type": "application/json"}) self.header_setup(headers={"Content-type": "application/json"})
c = process_tuples(w) c = process_tuples(w)
print(w, c, sep="\n") # print(w, c, sep="\n")
self.wfile.write(str(c).encode()) # self.wfile.write(str(c).encode())
# self.wfile.write(str(anotherone2).encode())
efg = funny_tuples_to_dicts_processor(anotherone2, c)
self.wfile.write(str(efg).encode())
return return
else: else:
with connectio.cursor(buffered=True) as cur: with connectio.cursor(buffered=True) as cur:
try: try:
res = cur.execute("select neme, kewod from tbale") res = cur.execute("select neme, kewod, cod from tbale")
wd = cur.fetchall() wd = cur.fetchall()
except Exception as e: except Exception as e:
print(e) print(e)
return # return ealy no else return # return ealy no else
# print(wd) # print(wd)
for a, v in wd: for a, v, c in wd:
# print(a,v) # print(a,v)
# print(self.path, a) # print(self.path, a)
if self.path.strip("/") == a: if self.path.lstrip("/") == a:
self.header_setup(200) if c == 200:
self.header_setup(c)
self.wfile.write(v.encode()) self.wfile.write(v.encode())
elif c in [301, 302]:
self.header_setup(c, {"Location": v})
return return
self.header_setup(404) self.header_setup(404)
self.wfile.write( self.wfile.write(
@@ -163,51 +231,6 @@ class ServerLogic(qqq.BaseHTTPRequestHandler):
) )
return 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): def do_POST(self):
self.youJustHaveToDoTheseStuff() self.youJustHaveToDoTheseStuff()
@@ -216,26 +239,31 @@ class ServerLogic(qqq.BaseHTTPRequestHandler):
return self.close() return self.close()
case "/endp": case "/endp":
self.header_setup() self.header_setup(headers={"Content-type": "text/plain"})
content_type = self.headers.get("Content-Type") content_type = self.headers.get("Content-Type")
content_length = int(self.headers["Content-Length"]) content_length = int(self.headers.get("Content-Length"))
with self.rfile as r: with self.rfile as r:
post_data_bytes = r.read(content_length) post_data_bytes = r.read(content_length)
## read how many bytes is the content length ## read how many bytes is the content length
self.wfile.write( self.wfile.write(
b"Ive got a POST at " b"Ive got a POST at "
+ self.headers["Origin"].encode() + self.headers.get("Origin", "nowhere lmao").encode()
+ b" from " + b" from "
+ self.headers["Referer"].encode() + self.headers.get("Referer", "who?").encode()
+ b" with the following data:\n" + b" with the following data:\n"
+ post_data_bytes + post_data_bytes
) )
if content_type == "application/json": if content_type == "application/json":
try: try:
data_dict: dict = json.loads(post_data_bytes.decode("utf-8")) data_dict: dict = json.loads(post_data_bytes)
except Exception as e: except Exception as e:
print(e) print(e)
self.wfile.write(
b"\nhaha your data sucks (json.load is strict/bad): "
+ str(e).encode()
)
return
(connectio, eror) = get_that_connection(connectionSettings) (connectio, eror) = get_that_connection(connectionSettings)
if eror: if eror:
print("coneti eror: ", eror) print("coneti eror: ", eror)
@@ -245,17 +273,47 @@ class ServerLogic(qqq.BaseHTTPRequestHandler):
) )
return return
print((*data_dict.keys(), *data_dict.values())) print((*data_dict.keys(), *data_dict.values()))
with connectio.cursor(buffered=True) as cur:
try:
anotherone = cur.execute(
"select column_name from INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'tbale'"
)
this_list_of_colnaes = cur.fetchall()
except Exception as e:
print("to day i cant get the damn colum")
this_list_of_colnaes = [u[0] for u in this_list_of_colnaes] # weird
if data_dict:
# tehwe: list = []
# for key in data_dict:
# if key in this_list_of_colnaes:
# tehwe.append(key)
wwwww = tuple(
key for key in data_dict if key in this_list_of_colnaes
)
#hmmm
# repated cod? huh?
# updated_Verybreakablequery_insert_into = (
# f"insert into tbale {wwwww} values {tuple('?'* len(wwwww))}"
updated_Verybreakablequery_insert_into = (
"insert into tbale ({0}) values ({0})".format(
",".join(["{}"] * len(wwwww))
).format(*wwwww, *("?",) * len(wwwww))
)
print(
tuple(data_dict[w] for w in wwwww),
updated_Verybreakablequery_insert_into,
sep="\n",
)
with connectio.cursor(prepared=True) as cur: with connectio.cursor(prepared=True) as cur:
try: try:
res = cur.execute( res = cur.execute(
"insert into tbale (kewod, expi) values (?, ?);", updated_Verybreakablequery_insert_into,
( tuple(data_dict[w] for w in wwwww),
data_dict["kewod"],
data_dict["expi"],
),
multi=True, multi=True,
) )
w2f = cur.fetchall()
print(res) print(res)
except Exception as e: except Exception as e:
print("execute err: ", e) print("execute err: ", e)
@@ -263,10 +321,13 @@ class ServerLogic(qqq.BaseHTTPRequestHandler):
b"\n is it very much Invalid: " + str(e).encode() b"\n is it very much Invalid: " + str(e).encode()
) )
else: else:
print(cur) # print(cur)
cur.execute("select * from tbale") # cur.execute("select * from tbale")
# aq142 = cur.fetchall()
self.wfile.write(b"\n We have this in the this: ") self.wfile.write(b"\n We have this in the this: ")
self.do_GET() # self.do_GET()
self.wfile.write(str(w2f).encode())
## WHAT THE FUCK IS THIS PART
else: else:
self.wfile.write(b"\nits of the type: ") self.wfile.write(b"\nits of the type: ")
self.wfile.write(content_type.encode()) self.wfile.write(content_type.encode())
+17 -5
View File
@@ -1,7 +1,19 @@
<form action="/endp" method="post"> <form action="/endp" method="post">
<label for="oneid"></label><input type="text" name="one" id="oneid"> <style>
<label for="twoid"></label><textarea name="two" id="twoid"></textarea> @scope {
<label for="threeid"></label><input type="number" name="three" id="threeid"> :not(style) {
<label for="fourid"></label><input type="number" name="four" id="fourid"> display: block;
<input type="submit" value="Sumit ur font assfuck"> }
p {
font-size: larger;
}
}
</style>
<label for="oneid">thing:<input type="text" name="one" id="oneid" /></label>
<label for="twoid">other:<textarea rows="7" cols="40" name="two" id="twoid"></textarea>
</label>
<label for="threeid">number:<input type="number" name="three" id="threeid" /></label>
<label for="fourid">probably will check with js idfk<input type="number" name="four" id="fourid" /></label>
<input type="submit" value="Sumit ur font assfuck" />
</form> </form>
+13 -1
View File
@@ -1,3 +1,15 @@
# ww # what by lda
its bad
## the most worstest server uve ever seen
i hate this more than you do i hate this more than you do
## you dont read the code
definition of a black box. things go in things go out, and what is inside stays unknown
## how do i run
you dont
+44 -12
View File
@@ -2,13 +2,14 @@ from typing import Callable, Dict, Any, List, Tuple, Union
from itertools import zip_longest from itertools import zip_longest
import datetime import datetime
class FunnyStuff: class FunnyStuff:
@staticmethod @staticmethod
def funny( def funny(
f: Callable, f: Callable,
# m: Dict["args": list(), "kwargs": dict()] # m: Dict["args": list(), "kwargs": dict()]
*a, *a,
**b **b,
): ):
# a = m.get("args", []) # a = m.get("args", [])
# b = m.get("kwargs", {}) # b = m.get("kwargs", {})
@@ -50,7 +51,7 @@ def function_splitter(
{}, {},
), ),
*e, *e,
**ae **ae,
): ):
"""inp: (a, (b)?, {c}?), *oa?, **okwa?, """inp: (a, (b)?, {c}?), *oa?, **okwa?,
exec wrapper if len args == 1 + args type str, exec wrapper if len args == 1 + args type str,
@@ -75,13 +76,13 @@ def function_splitter(
def funny_trycatch_wrapper( def funny_trycatch_wrapper(
tries: list[tuple[Callable| str, tuple[Any, ...] | Any, Dict[str, Any]]], tries: list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]],
excepts: dict[ excepts: dict[
BaseException, BaseException,
list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]], list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]],
] = {BaseException: [(lambda x: print(f"Error: {x}"))]}, ] = {BaseException: [(lambda x: print(f"Error: {x}"))]},
elses: list[tuple[Callable| str, tuple[Any, ...] | Any, Dict[str, Any]]] = [], elses: list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]] = [],
finallies: 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. Try, except, else, and finally wrapper for multiple functions or exec strings.
@@ -90,24 +91,42 @@ def funny_trycatch_wrapper(
:param excepts: Dictionary mapping exceptions to lists of functions or strings. :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 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. :param finallies: List of functions or strings to run after the try/except/else blocks.
:return:
superreturn = {
"rtries": [],
"rexcepts": {},
"relses": [],
"rfinallies": [],
}
""" """
superreturn = {
"rtries": [],
"rexcepts": {},
"relses": [],
"rfinallies": [],
}
try: try:
for oneTry in tries: for oneTry in tries:
function_splitter(oneTry) res = function_splitter(oneTry)
superreturn["rtries"].append(res)
except BaseException as e: except BaseException as e:
for exception_type, handlers in excepts.items(): for exception_type, handlers in excepts.items():
if isinstance(e, exception_type): if isinstance(e, exception_type):
for handler in handlers: for handler in handlers:
function_splitter(handler, e) f = function_splitter(handler, e) # do i give the thing the thng or the thing
superreturn["rexcepts"][handler] = res
break break
else: else:
for oneThing in elses: for oneThing in elses:
function_splitter(oneThing) res = function_splitter(oneThing)
superreturn["relses"].append(res)
finally: finally:
for oneThing in finallies: for oneThing in finallies:
function_splitter(oneThing) res = function_splitter(oneThing)
superreturn["rfinallies"].append(res)
return superreturn # lmao
thisone = "this string" thisone = "this string"
@@ -143,8 +162,9 @@ def funny_singular_Simple_trycatch(Onetry: tuple):
return (this, er) return (this, er)
def process_tuples(
def process_tuples(tuples: List[Tuple[Any, ...]], action: str = "timestamp") -> List[Tuple[Any, ...]]: tuples: List[Tuple[Any, ...]], action: str = "timestamp"
) -> List[Tuple[Any, ...]]:
""" """
Parse through a list of tuples and perform an action on datetime objects. Parse through a list of tuples and perform an action on datetime objects.
@@ -171,3 +191,15 @@ def process_tuples(tuples: List[Tuple[Any, ...]], action: str = "timestamp") ->
processed_tuples.append(tuple(processed_tuple)) processed_tuples.append(tuple(processed_tuple))
return processed_tuples return processed_tuples
def funny_tuples_to_dicts_processor(colname: list[tuple], coldata: list[tuple]):
if len(colname) != len(coldata[0]):
raise ValueError("unless im stupid, even then curate ur list first")
new_dicts: list[dict] = []
for ee in coldata:
new_dict = {}
for i, a in enumerate(ee):
new_dict[colname[i][0]] = a
new_dicts.append(new_dict)
return new_dicts