Files
wehub-resource-sync bcbd1bdb22
Integ / changes (push) Has been skipped
Pre-commit / pre-commit (push) Failing after 1s
CLI exit codes / changes (push) Has been skipped
Test (Install) / changes (push) Has been skipped
Test (Python) / changes (push) Has been skipped
Test (TypeScript) / changes (push) Has been skipped
CLI exit codes / cli-gate (push) Has been cancelled
Test (Install) / test-install-gate (push) Has been cancelled
Integ / integ-gate (push) Has been cancelled
Test (Python) / test-python-gate (push) Has been cancelled
Test (TypeScript) / test-typescript-gate (push) Has been cancelled
Test (Install) / python-minimal (3.12) (push) Has been cancelled
Test (Install) / python-minimal (3.11) (push) Has been cancelled
Test (Install) / python-extra (agno, mirage.agents.agno) (push) Has been cancelled
Test (Install) / python-extra (chroma, mirage.resource.chroma) (push) Has been cancelled
Test (Install) / python-extra (pdf, mirage.core.filetype.pdf) (push) Has been cancelled
Integ / integ (push) Has been cancelled
Integ / integ-database (push) Has been cancelled
Integ / integ-database-ts (push) Has been cancelled
Integ / integ-data (push) Has been cancelled
Integ / integ-ssh (push) Has been cancelled
Integ / integ-ssh-ts (push) Has been cancelled
Test (Python) / audit (push) Has been cancelled
Test (TypeScript) / test (push) Has been cancelled
Test (TypeScript) / python-fs-shim (push) Has been cancelled
CLI exit codes / Python CLI (push) Has been cancelled
CLI exit codes / TypeScript CLI (push) Has been cancelled
CLI exit codes / Cross-language snapshot interop (push) Has been cancelled
Test (Python) / test (push) Has been cancelled
Test (Python) / import-isolation (deepagents, openai, mirage.agents.openai_agents) (push) Has been cancelled
Test (Python) / import-isolation (deepagents, pydantic-ai, mirage.agents.pydantic_ai) (push) Has been cancelled
Integ / integ-ts (push) Has been cancelled
Integ / integ-fuse (push) Has been cancelled
Test (Install) / python-extra (databricks, mirage.resource.databricks_volume) (push) Has been cancelled
Test (Install) / python-extra (deepagents, mirage.agents.langchain) (push) Has been cancelled
Test (Install) / python-extra (email, mirage.resource.email) (push) Has been cancelled
Test (Install) / python-extra (fuse, mirage.fuse.mount) (push) Has been cancelled
Test (Install) / python-extra (hdf5, mirage.core.filetype.hdf5) (push) Has been cancelled
Test (Install) / python-extra (hf, mirage.resource.hf_buckets) (push) Has been cancelled
Test (Install) / python-extra (lancedb, mirage.resource.lancedb) (push) Has been cancelled
Test (Install) / python-extra (langfuse, mirage.resource.langfuse) (push) Has been cancelled
Test (Install) / python-extra (mongodb, mirage.resource.mongodb) (push) Has been cancelled
Test (Install) / python-extra (nextcloud, mirage.resource.nextcloud) (push) Has been cancelled
Test (Install) / python-extra (openai, mirage.agents.openai_agents) (push) Has been cancelled
Test (Install) / python-extra (openhands, mirage.agents.openhands, 3.12) (push) Has been cancelled
Test (Install) / python-extra (parquet, mirage.core.filetype.parquet) (push) Has been cancelled
Test (Install) / python-extra (postgres, mirage.resource.postgres) (push) Has been cancelled
Test (Install) / python-extra (pydantic-ai, mirage.agents.pydantic_ai) (push) Has been cancelled
Test (Install) / python-extra (qdrant, mirage.resource.qdrant) (push) Has been cancelled
Test (Install) / python-extra (redis, mirage.resource.redis) (push) Has been cancelled
Test (Install) / python-extra (s3, mirage.resource.s3) (push) Has been cancelled
Test (Install) / python-extra (ssh, mirage.resource.ssh) (push) Has been cancelled
Test (Install) / ts-minimal (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:30:44 +08:00

219 lines
6.2 KiB
Python

# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
from dataclasses import dataclass, field
from mirage.commands.builtin.find_eval import (And, Empty, Name, Not, Or, Path,
PredNode, TrueNode, Type)
from mirage.commands.builtin.find_helper import _parse_mtime, _parse_size
from mirage.commands.errors import FindParseError
_VALUE_PREDICATES = frozenset({
"-name",
"-iname",
"-path",
"-type",
"-size",
"-mtime",
"-maxdepth",
"-mindepth",
})
_BARE_PREDICATES = frozenset({
"-empty",
"-print",
"-print0",
"-delete",
"-ls",
"-depth",
})
_OPERATORS = frozenset({
"-not",
"!",
"-o",
"-or",
"-a",
"-and",
"(",
")",
})
_EXPRESSION_TOKENS = _VALUE_PREDICATES | _BARE_PREDICATES | _OPERATORS
_VALID_TYPES = frozenset({"b", "c", "d", "p", "f", "l", "s"})
_MAX_DEPTH = 100
@dataclass
class FindExpr:
tree: PredNode
maxdepth: int | None = None
mindepth: int | None = None
min_size: int | None = None
max_size: int | None = None
mtime_min: float | None = None
mtime_max: float | None = None
uses_empty: bool = False
@dataclass
class _State:
tokens: list[str]
pos: int = 0
depth: int = 0
expr: FindExpr = field(default_factory=lambda: FindExpr(tree=TrueNode()))
def _peek(state: _State) -> str | None:
return state.tokens[state.pos] if state.pos < len(state.tokens) else None
def _advance(state: _State) -> str | None:
tok = _peek(state)
if tok is not None:
state.pos += 1
return tok
def _type_node(value: str) -> Type:
if value in ("f", "file"):
return Type("f")
if value in ("d", "directory"):
return Type("d")
if value in _VALID_TYPES:
return Type(value)
raise FindParseError(f"find: Unknown argument to -type: {value}")
def _int_arg(value: str, flag: str) -> int:
try:
return int(value)
except ValueError as exc:
raise FindParseError(
f"find: invalid argument '{value}' to '{flag}'") from exc
def _size_arg(value: str) -> tuple[int | None, int | None]:
try:
return _parse_size(value)
except (ValueError, IndexError) as exc:
raise FindParseError(
f"find: invalid argument '{value}' to '-size'") from exc
def _mtime_arg(value: str) -> tuple[float | None, float | None]:
try:
return _parse_mtime(value)
except (ValueError, IndexError) as exc:
raise FindParseError(
f"find: invalid argument '{value}' to '-mtime'") from exc
def _parse_primary(state: _State) -> PredNode:
tok = _advance(state)
if tok is None:
raise FindParseError("find: expected predicate")
if tok in _VALUE_PREDICATES:
value = _advance(state)
if value is None:
raise FindParseError(f"find: missing argument to '{tok}'")
if tok == "-name":
return Name(value)
if tok == "-iname":
return Name(value, icase=True)
if tok == "-path":
return Path(value)
if tok == "-type":
return _type_node(value)
if tok == "-maxdepth":
state.expr.maxdepth = _int_arg(value, "-maxdepth")
return TrueNode()
if tok == "-mindepth":
state.expr.mindepth = _int_arg(value, "-mindepth")
return TrueNode()
if tok == "-size":
state.expr.min_size, state.expr.max_size = _size_arg(value)
return TrueNode()
state.expr.mtime_min, state.expr.mtime_max = _mtime_arg(value)
return TrueNode()
if tok == "-empty":
state.expr.uses_empty = True
return Empty()
if tok in _BARE_PREDICATES:
return TrueNode()
raise FindParseError(f"find: unknown predicate '{tok}'")
def _parse_factor(state: _State) -> PredNode:
state.depth += 1
if state.depth > _MAX_DEPTH:
raise FindParseError("find: expression too deeply nested")
try:
tok = _peek(state)
if tok in ("-not", "!"):
_advance(state)
return Not(_parse_factor(state))
if tok == "(":
_advance(state)
node = _parse_or(state)
if _peek(state) != ")":
raise FindParseError("find: unbalanced parentheses")
_advance(state)
return node
return _parse_primary(state)
finally:
state.depth -= 1
def _parse_and(state: _State) -> PredNode:
factors = [_parse_factor(state)]
while True:
tok = _peek(state)
if tok in ("-a", "-and"):
_advance(state)
factors.append(_parse_factor(state))
continue
if tok is None or tok in ("-o", "-or", ")"):
break
factors.append(_parse_factor(state))
return factors[0] if len(factors) == 1 else And(factors)
def _parse_or(state: _State) -> PredNode:
terms = [_parse_and(state)]
while _peek(state) in ("-o", "-or"):
_advance(state)
terms.append(_parse_and(state))
return terms[0] if len(terms) == 1 else Or(terms)
def find_expr_tail(raw_argv: list[str]) -> list[str]:
for i, tok in enumerate(raw_argv):
if tok in _EXPRESSION_TOKENS or (tok.startswith("-") and len(tok) > 1):
return raw_argv[i:]
return []
def parse_find_expression(tokens: list[str]) -> FindExpr:
if not tokens:
return FindExpr(tree=TrueNode())
state = _State(tokens=tokens)
tree = _parse_or(state)
if _peek(state) is not None:
raise FindParseError(f"find: unexpected token '{_peek(state)}'")
state.expr.tree = tree
return state.expr