Файловый менеджер - Редактировать - /usr/lib/python3/dist-packages/uaclient/apt_news.py
Назад
import datetime import json import logging import os import unicodedata from typing import List, Optional import apt_pkg from uaclient import defaults, messages, system, util from uaclient.clouds.identity import get_cloud_type from uaclient.config import UAConfig from uaclient.data_types import ( BoolDataValue, DataObject, DatetimeDataValue, Field, StringDataValue, data_list, ) from uaclient.files import state_files from uaclient.jobs.update_messaging import ( ContractExpiryStatus, get_contract_expiry_status, ) class AptNewsMessageSelectors(DataObject): fields = [ Field("codenames", data_list(StringDataValue), required=False), Field("clouds", data_list(StringDataValue), required=False), Field("pro", BoolDataValue, required=False), ] def __init__( self, *, codenames: Optional[List[str]] = None, clouds: Optional[List[str]] = None, pro: Optional[bool] = None ): self.codenames = codenames self.clouds = clouds self.pro = pro class AptNewsMessage(DataObject): fields = [ Field("begin", DatetimeDataValue), Field("end", DatetimeDataValue, required=False), Field("selectors", AptNewsMessageSelectors, required=False), Field("lines", data_list(StringDataValue)), ] def __init__( self, *, begin: datetime.datetime, end: Optional[datetime.datetime] = None, selectors: Optional[AptNewsMessageSelectors] = None, lines: List[str] ): self.begin = begin self.end = end self.selectors = selectors self.lines = lines def format_message(msg: AptNewsMessage) -> str: result = "#\n" for line in msg.lines: result += "# {}\n".format(line) result += "#\n" return result def do_selectors_apply( cfg: UAConfig, selectors: Optional[AptNewsMessageSelectors] ) -> bool: if selectors is None: return True if selectors.codenames is not None: if system.get_platform_info()["series"] not in selectors.codenames: return False if selectors.clouds is not None: cloud_id, fail = get_cloud_type() if fail is not None: return False if cloud_id not in selectors.clouds: return False if selectors.pro is not None: if selectors.pro != cfg.is_attached: return False return True def do_dates_apply( begin: datetime.datetime, end: Optional[datetime.datetime] ) -> bool: now = datetime.datetime.now(datetime.timezone.utc) if now < begin: return False one_month_after_begin = begin + datetime.timedelta(days=30) if end is None or end > one_month_after_begin: end_to_use = one_month_after_begin else: end_to_use = end if now > end_to_use: return False return True def is_control_char(c: str) -> bool: return unicodedata.category(c)[0] == "C" def is_message_valid(msg: AptNewsMessage) -> bool: if len(msg.lines) < 1: return False if len(msg.lines) > 3: return False for line in msg.lines: if any([is_control_char(c) for c in line]): return False if len(line) > 77: return False return True def select_message( cfg: UAConfig, messages: List[dict] ) -> Optional[AptNewsMessage]: for msg_dict in messages: try: msg = AptNewsMessage.from_dict(msg_dict) except Exception as e: logging.debug("msg failed parsing: %r", e) continue if not is_message_valid(msg): logging.debug("msg not valid: %r", msg) continue if not do_dates_apply(msg.begin, msg.end): logging.debug("msg dates don't apply: %r", msg) continue if not do_selectors_apply(cfg, msg.selectors): logging.debug("msg selectors don't apply: %r", msg) continue return msg return None def fetch_aptnews_json(cfg: UAConfig): os.makedirs(defaults.UAC_RUN_PATH, exist_ok=True) acq = apt_pkg.Acquire() apt_news_file = apt_pkg.AcquireFile( acq, cfg.apt_news_url, destdir=defaults.UAC_RUN_PATH ) acq.run() apt_news_contents = system.load_file(apt_news_file.destfile) return json.loads( apt_news_contents, cls=util.DatetimeAwareJSONDecoder, ) def fetch_and_process_apt_news(cfg: UAConfig): try: news_dict = fetch_aptnews_json(cfg) msg = select_message(cfg, news_dict.get("messages", [])) logging.debug("using msg: %r", msg) if msg is not None: msg_str = format_message(msg) state_files.apt_news_contents_file.write(msg_str) else: state_files.apt_news_contents_file.delete() except Exception as e: logging.debug("something went wrong while processing apt_news: %r", e) state_files.apt_news_contents_file.delete() def local_apt_news(cfg: UAConfig) -> bool: """ :return: True if local news was written, False otherwise """ expiry_status, remaining_days = get_contract_expiry_status(cfg) if expiry_status == ContractExpiryStatus.ACTIVE_EXPIRED_SOON: state_files.apt_news_contents_file.write( messages.CONTRACT_EXPIRES_SOON_APT_NEWS.format( remaining_days=remaining_days ) ) return True if expiry_status == ContractExpiryStatus.EXPIRED_GRACE_PERIOD: grace_period_remaining = ( defaults.CONTRACT_EXPIRY_GRACE_PERIOD_DAYS + remaining_days ) exp_dt = cfg.machine_token_file.contract_expiry_datetime if exp_dt is None: exp_dt_str = "Unknown" else: exp_dt_str = exp_dt.strftime("%d %b %Y") state_files.apt_news_contents_file.write( messages.CONTRACT_EXPIRED_GRACE_PERIOD_APT_NEWS.format( expired_date=exp_dt_str, remaining_days=grace_period_remaining ) ) return True if expiry_status == ContractExpiryStatus.EXPIRED: state_files.apt_news_contents_file.write( messages.CONTRACT_EXPIRED_APT_NEWS ) return True return False def update_apt_news(cfg: UAConfig): local_news_written = local_apt_news(cfg) if not local_news_written: apt_pkg.init() fetch_and_process_apt_news(cfg)
| ver. 1.4 |
Github
|
.
| PHP 7.4.3-4ubuntu2.28 | Генерация страницы: 0 |
proxy
|
phpinfo
|
Настройка