Elgato_dark/bunkerweb
1
0
Fork 0
mirror of https://github.com/bunkerity/bunkerweb synced 2026-05-24 09:28:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Refactor get_jobs_cache_files method and add cache content to cache files back in the web UI

Browse source
This commit is contained in:
Théophile Diot 2024-03-16 11:25:38 +00:00
parent d1077156c9
commit bfd4eb201d
No known key found for this signature in database
GPG key ID: 248FEA4BAE400D06
4 changed files with 69 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
src/common/db/Database.py
View file

@ -1753,42 +1753,47 @@ class Database:
ret_data["data"] = data.data
return ret_data
def get_jobs_cache_files(self, *, job_name: str = "", plugin_id: str = "", with_data: bool = False) -> List[Dict[str, Any]]:
def get_jobs_cache_files(self, *, job_name: str = "", plugin_id: str = "") -> List[Dict[str, Any]]:
"""Get jobs cache files."""
with self.__db_session() as session:
entities = [Jobs_cache.job_name, Jobs_cache.service_id, Jobs_cache.file_name]
if with_data:
entities.append(Jobs_cache.data)
query = session.query(Jobs_cache).with_entities(*entities)
filters = {}
query = session.query(Jobs_cache).with_entities(Jobs_cache.job_name, Jobs_cache.service_id, Jobs_cache.file_name, Jobs_cache.data)
if job_name:
query = query.filter_by(job_name=job_name)
filters["name"] = job_name
db_cache = query.all()
if not db_cache:
return []
job_names = []
if plugin_id:
filters = {"plugin_id": plugin_id}
if job_name:
filters["name"] = job_name
job_names = [job.name for job in session.query(Jobs).with_entities(Jobs.name).filter_by(**filters)]
if not job_names:
return []
filters["plugin_id"] = plugin_id
query = session.query(Jobs).with_entities(Jobs.name, Jobs.plugin_id)
if filters:
query = query.filter_by(**filters)
jobs = {}
for job in query:
jobs[job.name] = job.plugin_id
if not jobs:
return []
cache_files = []
for cache in db_cache:
if job_name and cache.job_name not in job_names:
if cache.job_name not in jobs:
continue
cache_files.append(
{
"plugin_id": jobs[cache.job_name],
"job_name": cache.job_name,
"service_id": cache.service_id,
"file_name": cache.file_name,
"data": "Download file to view content" if not with_data else cache.data,
"data": cache.data,
}
)
return cache_files

2
src/common/utils/jobs.py
View file

@ -54,7 +54,7 @@ class Job:
"""Restore job cache files from database."""
ret = True
with LOCK:
job_cache_files = self.db.get_jobs_cache_files(plugin_id=plugin_id or self.job_path.name, with_data=True) # type: ignore
job_cache_files = self.db.get_jobs_cache_files(plugin_id=plugin_id or self.job_path.name) # type: ignore
job_name = job_name or self.job_name
plugin_cache_files = set()

4
src/ui/Dockerfile
View file

@ -38,8 +38,8 @@ COPY --from=builder --chown=0:101 /usr/share/bunkerweb /usr/share/bunkerweb
WORKDIR /usr/share/bunkerweb
# Add ui user, drop bwcli, install runtime dependencies, create data folders and set permissions
RUN apk add --no-cache bash && \
# Add ui user, install runtime dependencies, create data folders and set permissions
RUN apk add --no-cache bash libmagic && \
addgroup -g 101 ui && \
adduser -h /var/cache/nginx -g ui -s /bin/sh -G ui -D -H -u 101 ui && \
echo "Docker" > INTEGRATION && \

63
src/ui/utils.py
View file

@ -5,6 +5,7 @@ from io import BytesIO
from os.path import join
from typing import List, Optional
from magic import Magic
from qrcode.main import QRCode
@ -131,40 +132,68 @@ def path_to_dict(
"can_create_folders": False,
"can_edit": False,
"can_delete": False,
"children": [
{
"name": service,
"type": "folder",
"path": join(path, service),
"can_create_files": False,
"can_create_folders": False,
"can_edit": False,
"can_delete": False,
"children": [],
}
for service in services
],
"children": [],
}
plugins = []
paths = []
for conf in db_data:
if conf["plugin_id"] not in plugins:
d["children"].append(
{
"name": conf["plugin_id"],
"type": "folder",
"path": join(path, conf["plugin_id"]),
"can_create_files": True,
"can_create_folders": False,
"can_edit": False,
"can_delete": False,
"children": [],
}
)
plugins.append(conf["plugin_id"])
paths.append(join(path, conf["plugin_id"]))
mime = Magic(mime=True)
file_type = mime.from_buffer(conf["data"])
print(join(path, conf["plugin_id"], conf["service_id"] or "", conf["file_name"]), file_type, flush=True)
file_info = {
"name": join(conf["job_name"], conf["file_name"]),
"name": conf["file_name"],
"type": "file",
"path": join(
path,
conf["plugin_id"],
conf["service_id"] or "",
conf["file_name"],
),
"can_edit": False,
"can_delete": False,
"can_download": True,
"content": conf["data"],
"content": "Download file to view content" if file_type != "text/plain" else conf["data"].decode("utf-8"),
}
if conf["service_id"]:
d["children"][[x["name"] for x in d["children"]].index(conf["service_id"])]["children"].append(file_info)
if join(conf["plugin_id"], conf["service_id"]) not in paths:
d["children"][[x["name"] for x in d["children"]].index(conf["plugin_id"])]["children"].append(
{
"name": conf["service_id"],
"type": "folder",
"path": join(path, conf["plugin_id"], conf["service_id"]),
"can_create_files": True,
"can_create_folders": False,
"can_edit": False,
"can_delete": False,
"children": [],
}
)
paths.append(join(conf["plugin_id"], conf["service_id"]))
data_plugin = d["children"][[x["name"] for x in d["children"]].index(conf["plugin_id"])]
data_plugin["children"][[x["name"] for x in data_plugin["children"]].index(conf["service_id"])]["children"].append(file_info)
else:
d["children"].append(file_info)
d["children"][[x["name"] for x in d["children"]].index(conf["plugin_id"])]["children"].append(file_info)
return d