__ __ __ __ _____ _ _ _____ _ _ _
| \/ | \ \ / / | __ \ (_) | | / ____| | | | |
| \ / |_ __\ V / | |__) | __ ___ ____ _| |_ ___ | (___ | |__ ___| | |
| |\/| | '__|> < | ___/ '__| \ \ / / _` | __/ _ \ \___ \| '_ \ / _ \ | |
| | | | |_ / . \ | | | | | |\ V / (_| | || __/ ____) | | | | __/ | |
|_| |_|_(_)_/ \_\ |_| |_| |_| \_/ \__,_|\__\___| |_____/|_| |_|\___V 2.1
if you need WebShell for Seo everyday contact me on Telegram
Telegram Address : @jackleet
For_More_Tools:
# -*- coding: utf-8 -*-
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2025 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#
"""Parsing and serialization functions for crontab entries."""
from .constants import (
CRON_ENV_ASSIGNMENT_PATTERN,
CRON_SCHEDULE_PATTERN,
WEBSITE_CRON_BEGIN_PATTERN,
WEBSITE_CRON_END_PATTERN,
)
from .structure import (
BeginWebsite,
CommentLine,
CrontabEntry,
CrontabStructure,
EndWebsite,
EnvAssignmentLine,
ParsedCrontabLine,
)
def parse_crontab_line(line: bytes) -> CrontabEntry:
"""
Parse a crontab line into a cron entry, comment, or website marker.
Args:
line: A single line from crontab output (bytes)
"""
stripped = line.strip()
# Check for website cron begin marker
begin_match = WEBSITE_CRON_BEGIN_PATTERN.match(stripped)
if begin_match:
docroot_bytes = begin_match.group(1).strip()
docroot = docroot_bytes.decode("utf-8", errors="replace")
return BeginWebsite(docroot=docroot)
# Check for website cron end marker
if WEBSITE_CRON_END_PATTERN.match(stripped):
return EndWebsite()
# Handle empty lines and comments
if not stripped:
return CommentLine(content=line)
if stripped.startswith(b"#"):
return CommentLine(content=line)
# Environment-assignment line (`name = value`) — must be classified
# distinctly from comments so the wrap step can drop these in per-site
# sections; otherwise crond honours them and a user-supplied `SHELL=`
# would run before the isolation wrapper for following job lines.
if CRON_ENV_ASSIGNMENT_PATTERN.match(stripped):
return EnvAssignmentLine(content=line)
# Try to match cron schedule + command
schedule_match = CRON_SCHEDULE_PATTERN.match(stripped)
if not schedule_match:
return CommentLine(content=line)
schedule = schedule_match.group(1)
command = schedule_match.group(2)
if line.endswith(b"\n"):
command += b"\n"
return ParsedCrontabLine(
schedule=schedule,
command=command,
)
def parse_crontab_structure(input_data: bytes) -> CrontabStructure:
"""
Parse crontab file into structured format with user records and docroot sections.
Supports both old format (with isolation wrapper prefixes) and new format
(with section markers).
Args:
input_data: bytes containing crontab entries
Returns:
CrontabStructure: Structure containing global_records and docroot_sections
"""
lines = input_data.splitlines(keepends=True)
crontab_records: dict[
str | None,
list[CrontabEntry],
] = {}
current_docroot: str | None = None
for line in lines:
parsed = parse_crontab_line(line)
if isinstance(parsed, BeginWebsite):
current_docroot = parsed.docroot
continue
if isinstance(parsed, EndWebsite):
current_docroot = None
continue
crontab_records.setdefault(current_docroot, []).append(parsed)
global_records = crontab_records.pop(None, [])
return CrontabStructure(
global_records=global_records,
docroot_sections=crontab_records,
)
def write_crontab_structure(structure: CrontabStructure) -> bytes:
"""
Write crontab structure to bytes in new format.
Format:
- User records at the top
- Then docroot sections with markers:
## WEBSITE CRON BEGIN /path/to/docroot
... entries ...
## WEBSITE CRON END
Args:
structure: CrontabStructure with user_records and docroot_sections
Returns:
bytes: Crontab content in new format
"""
result_parts = entries_to_str_list(structure.global_records)
# Add docroot sections (preserve insertion order)
for docroot in structure.docroot_sections.keys():
entries = structure.docroot_sections[docroot]
if not entries:
continue
if '\n' in docroot or '\r' in docroot:
raise ValueError(f'Invalid docroot: {docroot!r}')
result_parts.append(f"## WEBSITE CRON BEGIN {docroot}\n".encode("utf-8"))
result_parts.extend(entries_to_str_list(entries))
result_parts.append(b"## WEBSITE CRON END\n")
return b"".join(result_parts)
def entries_to_str_list(entries: list[CrontabEntry], without_wrapper: bool = False) -> list[bytes]:
"""
Converts parsed crontab entries into a string list.
"""
lines = []
for entry in entries:
if isinstance(entry, ParsedCrontabLine):
line = (
entry.schedule
+ b" "
+ (entry.get_clean_command() if without_wrapper else entry.command)
)
if not line.endswith(b"\n"):
line += b"\n"
lines.append(line)
elif isinstance(entry, CommentLine):
lines.append(entry.content)
elif isinstance(entry, EnvAssignmentLine):
lines.append(entry.content)
return lines