Differential D6695 Diff 24481 swh/web/browse/snapshot_context.py

Changeset View

Standalone View

swh/web/browse/snapshot_context.py

Show All 16 Lines
from swh.model.swhids import CoreSWHID, ObjectType		from swh.model.swhids import CoreSWHID, ObjectType
from swh.web.browse.utils import (		from swh.web.browse.utils import (
format_log_entries,		format_log_entries,
gen_release_link,		gen_release_link,
gen_revision_link,		gen_revision_link,
gen_revision_log_link,		gen_revision_log_link,
gen_revision_url,		gen_revision_url,
gen_snapshot_link,		gen_snapshot_link,
get_directory_entries,
get_readme_to_display,
)		)
from swh.web.common import archive		from swh.web.common import archive
from swh.web.common.exc import BadInputExc, NotFoundExc, http_status_code_message		from swh.web.common.exc import BadInputExc, NotFoundExc
from swh.web.common.identifiers import get_swhids_info		from swh.web.common.identifiers import get_swhids_info
from swh.web.common.origin_visits import get_origin_visit		from swh.web.common.origin_visits import get_origin_visit
from swh.web.common.typing import (		from swh.web.common.typing import (
DirectoryMetadata,
OriginInfo,		OriginInfo,
SnapshotBranchInfo,		SnapshotBranchInfo,
SnapshotContext,		SnapshotContext,
SnapshotReleaseInfo,		SnapshotReleaseInfo,
SWHObjectInfo,		SWHObjectInfo,
)		)
from swh.web.common.utils import (		from swh.web.common.utils import format_utc_iso_date, gen_path_info, reverse
format_utc_iso_date,
gen_path_info,
reverse,
swh_object_icons,
)
from swh.web.config import get_config		from swh.web.config import get_config

_empty_snapshot_id = Snapshot(branches={}).id.hex()		_empty_snapshot_id = Snapshot(branches={}).id.hex()


def _get_branch(branches, branch_name, snapshot_id):		def _get_branch(branches, branch_name, snapshot_id):
"""		"""
Utility function to get a specific branch from a snapshot.		Utility function to get a specific branch from a snapshot.
▲ Show 20 Lines • Show All 656 Lines • ▼ Show 20 Lines	def _build_breadcrumbs(snapshot_context: SnapshotContext, path: str):
return breadcrumbs		return breadcrumbs


def _check_origin_url(snapshot_id, origin_url):		def _check_origin_url(snapshot_id, origin_url):
if snapshot_id is None and origin_url is None:		if snapshot_id is None and origin_url is None:
raise BadInputExc("An origin URL must be provided as query parameter.")		raise BadInputExc("An origin URL must be provided as query parameter.")


def browse_snapshot_directory(		# def browse_snapshot_directory(
request, snapshot_id=None, origin_url=None, timestamp=None, path=None		# request, snapshot_id=None, origin_url=None, timestamp=None, path=None
):		# ):
"""		# """
Django view implementation for browsing a directory in a snapshot context.		# Django view implementation for browsing a directory in a snapshot context.
"""		# """
_check_origin_url(snapshot_id, origin_url)		# _check_origin_url(snapshot_id, origin_url)

snapshot_context = get_snapshot_context(		# snapshot_context = get_snapshot_context(
snapshot_id=snapshot_id,		# snapshot_id=snapshot_id,
origin_url=origin_url,		# origin_url=origin_url,
timestamp=timestamp,		# timestamp=timestamp,
visit_id=request.GET.get("visit_id"),		# visit_id=request.GET.get("visit_id"),
path=path,		# path=path,
browse_context="directory",		# browse_context="directory",
branch_name=request.GET.get("branch"),		# branch_name=request.GET.get("branch"),
release_name=request.GET.get("release"),		# release_name=request.GET.get("release"),
revision_id=request.GET.get("revision"),		# revision_id=request.GET.get("revision"),
)		# )

root_directory = snapshot_context["root_directory"]		# root_directory = snapshot_context["root_directory"]
sha1_git = root_directory		# sha1_git = root_directory
error_info = {		# error_info = {
"status_code": 200,		# "status_code": 200,
"description": None,		# "description": None,
}		# }
if root_directory and path:		# if root_directory and path:
try:		# try:
dir_info = archive.lookup_directory_with_path(root_directory, path)		# dir_info = archive.lookup_directory_with_path(root_directory, path)
sha1_git = dir_info["target"]		# sha1_git = dir_info["target"]
except NotFoundExc as e:		# except NotFoundExc as e:
sha1_git = None		# sha1_git = None
error_info["status_code"] = 404		# error_info["status_code"] = 404
error_info["description"] = f"NotFoundExc: {str(e)}"		# error_info["description"] = f"NotFoundExc: {str(e)}"

dirs = []		# dirs = []
files = []		# files = []
if sha1_git:		# if sha1_git:
dirs, files = get_directory_entries(sha1_git)		# dirs, files = get_directory_entries(sha1_git)

origin_info = snapshot_context["origin_info"]		# origin_info = snapshot_context["origin_info"]
visit_info = snapshot_context["visit_info"]		# visit_info = snapshot_context["visit_info"]
url_args = snapshot_context["url_args"]		# url_args = snapshot_context["url_args"]
query_params = dict(snapshot_context["query_params"])		# query_params = dict(snapshot_context["query_params"])
revision_id = snapshot_context["revision_id"]		# revision_id = snapshot_context["revision_id"]
snapshot_id = snapshot_context["snapshot_id"]		# snapshot_id = snapshot_context["snapshot_id"]

if origin_info:		# if origin_info:
browse_view_name = "browse-origin-directory"		# browse_view_name = "browse-origin-directory"
else:		# else:
browse_view_name = "browse-snapshot-directory"		# browse_view_name = "browse-snapshot-directory"

breadcrumbs = _build_breadcrumbs(snapshot_context, path)		# breadcrumbs = _build_breadcrumbs(snapshot_context, path)

path = "" if path is None else (path + "/")		# path = "" if path is None else (path + "/")

for d in dirs:		# for d in dirs:
if d["type"] == "rev":		# if d["type"] == "rev":
d["url"] = reverse("browse-revision", url_args={"sha1_git": d["target"]})		# d["url"] = reverse("browse-revision", url_args={"sha1_git": d["target"]})
else:		# else:
query_params["path"] = path + d["name"]		# query_params["path"] = path + d["name"]
d["url"] = reverse(		# d["url"] = reverse(
browse_view_name, url_args=url_args, query_params=query_params		# browse_view_name, url_args=url_args, query_params=query_params
)		# )

sum_file_sizes = 0		# sum_file_sizes = 0

readmes = {}		# readmes = {}

if origin_info:		# if origin_info:
browse_view_name = "browse-origin-content"		# browse_view_name = "browse-origin-content"
else:		# else:
browse_view_name = "browse-snapshot-content"		# browse_view_name = "browse-snapshot-content"

for f in files:		# for f in files:
query_params["path"] = path + f["name"]		# query_params["path"] = path + f["name"]
f["url"] = reverse(		# f["url"] = reverse(
browse_view_name, url_args=url_args, query_params=query_params		# browse_view_name, url_args=url_args, query_params=query_params
)		# )
if f["length"] is not None:		# if f["length"] is not None:
sum_file_sizes += f["length"]		# sum_file_sizes += f["length"]
if f["name"].lower().startswith("readme"):		# if f["name"].lower().startswith("readme"):
readmes[f["name"]] = f["checksums"]["sha1"]		# readmes[f["name"]] = f["checksums"]["sha1"]

readme_name, readme_url, readme_html = get_readme_to_display(readmes)		# readme_name, readme_url, readme_html = get_readme_to_display(readmes)

if origin_info:		# if origin_info:
browse_view_name = "browse-origin-log"		# browse_view_name = "browse-origin-log"
else:		# else:
browse_view_name = "browse-snapshot-log"		# browse_view_name = "browse-snapshot-log"

history_url = None		# history_url = None
if snapshot_id != _empty_snapshot_id:		# if snapshot_id != _empty_snapshot_id:
query_params.pop("path", None)		# query_params.pop("path", None)
history_url = reverse(		# history_url = reverse(
browse_view_name, url_args=url_args, query_params=query_params		# browse_view_name, url_args=url_args, query_params=query_params
)		# )

nb_files = None		# nb_files = None
nb_dirs = None		# nb_dirs = None
dir_path = None		# dir_path = None
if root_directory:		# if root_directory:
nb_files = len(files)		# nb_files = len(files)
nb_dirs = len(dirs)		# nb_dirs = len(dirs)
dir_path = "/" + path		# dir_path = "/" + path

swh_objects = []		# swh_objects = []
vault_cooking = {}		# vault_cooking = {}
revision_found = True		# revision_found = True

if sha1_git is None and revision_id is not None:		# if sha1_git is None and revision_id is not None:
try:		# try:
archive.lookup_revision(revision_id)		# archive.lookup_revision(revision_id)
except NotFoundExc:		# except NotFoundExc:
revision_found = False		# revision_found = False

if sha1_git is not None:		# if sha1_git is not None:
swh_objects.append(		# swh_objects.append(
SWHObjectInfo(object_type=ObjectType.DIRECTORY, object_id=sha1_git)		# SWHObjectInfo(object_type=ObjectType.DIRECTORY, object_id=sha1_git)
)		# )
vault_cooking.update(		# vault_cooking.update(
{"directory_context": True, "directory_swhid": f"swh:1:dir:{sha1_git}",}		# {"directory_context": True, "directory_swhid": f"swh:1:dir:{sha1_git}",}
)		# )
if revision_id is not None and revision_found:		# if revision_id is not None and revision_found:
swh_objects.append(		# swh_objects.append(
SWHObjectInfo(object_type=ObjectType.REVISION, object_id=revision_id)		# SWHObjectInfo(object_type=ObjectType.REVISION, object_id=revision_id)
)		# )
vault_cooking.update(		# vault_cooking.update(
{"revision_context": True, "revision_swhid": f"swh:1:rev:{revision_id}",}		# {"revision_context": True, "revision_swhid": f"swh:1:rev:{revision_id}",}
)		# )
swh_objects.append(		# swh_objects.append(
SWHObjectInfo(object_type=ObjectType.SNAPSHOT, object_id=snapshot_id)		# SWHObjectInfo(object_type=ObjectType.SNAPSHOT, object_id=snapshot_id)
)		# )

visit_date = None		# visit_date = None
visit_type = None		# visit_type = None
if visit_info:		# if visit_info:
visit_date = format_utc_iso_date(visit_info["date"])		# visit_date = format_utc_iso_date(visit_info["date"])
visit_type = visit_info["type"]		# visit_type = visit_info["type"]

release_id = snapshot_context["release_id"]		# release_id = snapshot_context["release_id"]
if release_id:		# if release_id:
swh_objects.append(		# swh_objects.append(
SWHObjectInfo(object_type=ObjectType.RELEASE, object_id=release_id)		# SWHObjectInfo(object_type=ObjectType.RELEASE, object_id=release_id)
)		# )

dir_metadata = DirectoryMetadata(		# dir_metadata = DirectoryMetadata(
object_type=ObjectType.DIRECTORY,		# object_type=ObjectType.DIRECTORY,
object_id=sha1_git,		# object_id=sha1_git,
directory=sha1_git,		# directory=sha1_git,
nb_files=nb_files,		# nb_files=nb_files,
nb_dirs=nb_dirs,		# nb_dirs=nb_dirs,
sum_file_sizes=sum_file_sizes,		# sum_file_sizes=sum_file_sizes,
root_directory=root_directory,		# root_directory=root_directory,
path=dir_path,		# path=dir_path,
revision=revision_id,		# revision=revision_id,
revision_found=revision_found,		# revision_found=revision_found,
release=release_id,		# release=release_id,
snapshot=snapshot_id,		# snapshot=snapshot_id,
origin_url=origin_url,		# origin_url=origin_url,
visit_date=visit_date,		# visit_date=visit_date,
visit_type=visit_type,		# visit_type=visit_type,
)		# )

swhids_info = get_swhids_info(swh_objects, snapshot_context, dir_metadata)		# swhids_info = get_swhids_info(swh_objects, snapshot_context, dir_metadata)

dir_path = "/".join([bc["name"] for bc in breadcrumbs]) + "/"		# dir_path = "/".join([bc["name"] for bc in breadcrumbs]) + "/"
context_found = "snapshot: %s" % snapshot_context["snapshot_id"]		# context_found = "snapshot: %s" % snapshot_context["snapshot_id"]
if origin_info:		# if origin_info:
context_found = "origin: %s" % origin_info["url"]		# context_found = "origin: %s" % origin_info["url"]
heading = "Directory - %s - %s - %s" % (		# heading = "Directory - %s - %s - %s" % (
dir_path,		# dir_path,
snapshot_context["branch"],		# snapshot_context["branch"],
context_found,		# context_found,
)		# )

top_right_link = None		# top_right_link = None
if not snapshot_context["is_empty"] and revision_found:		# if not snapshot_context["is_empty"] and revision_found:
top_right_link = {		# top_right_link = {
"url": history_url,		# "url": history_url,
"icon": swh_object_icons["revisions history"],		# "icon": swh_object_icons["revisions history"],
"text": "History",		# "text": "History",
}		# }

return render(		# return render(
request,		# request,
"browse/directory.html",		# "browse/directory.html",
{		# {
"heading": heading,		# "heading": heading,
"swh_object_name": "Directory",		# "swh_object_name": "Directory",
"swh_object_metadata": dir_metadata,		# "swh_object_metadata": dir_metadata,
"dirs": dirs,		# "dirs": dirs,
"files": files,		# "files": files,
"breadcrumbs": breadcrumbs if root_directory else [],		# "breadcrumbs": breadcrumbs if root_directory else [],
"top_right_link": top_right_link,		# "top_right_link": top_right_link,
"readme_name": readme_name,		# "readme_name": readme_name,
"readme_url": readme_url,		# "readme_url": readme_url,
"readme_html": readme_html,		# "readme_html": readme_html,
"snapshot_context": snapshot_context,		# "snapshot_context": snapshot_context,
"vault_cooking": vault_cooking,		# "vault_cooking": vault_cooking,
"show_actions": True,		# "show_actions": True,
"swhids_info": swhids_info,		# "swhids_info": swhids_info,
"error_code": error_info["status_code"],		# "error_code": error_info["status_code"],
"error_message": http_status_code_message.get(error_info["status_code"]),		# "error_message": http_status_code_message.get(error_info["status_code"]),
"error_description": error_info["description"],		# "error_description": error_info["description"],
},		# },
status=error_info["status_code"],		# status=error_info["status_code"],
)		# )


PER_PAGE = 100		PER_PAGE = 100


def browse_snapshot_log(request, snapshot_id=None, origin_url=None, timestamp=None):		def browse_snapshot_log(request, snapshot_id=None, origin_url=None, timestamp=None):
"""		"""
Django view implementation for browsing a revision history in a		Django view implementation for browsing a revision history in a
▲ Show 20 Lines • Show All 392 Lines • Show Last 20 Lines