Write a spreadsheet with results for one query::

    query-report [year options] [-O OUTPUT.ods] <query string>

    query-report [year options] [-O OUTPUT.ods] < QUERY_FILE.bql

query-report also accepts all the same options as bean-query, like ``--format``

and ``--numberify``.

"""

# Copyright © 2021  Brett Smith

# License: AGPLv3-or-later WITH Beancount-Plugin-Additional-Permission-1.0

#

# Full copyright and licensing details can be found at toplevel file

# LICENSE.txt in the repository.

import argparse

import contextlib

import datetime

import enum

import functools

import itertools

import logging

import os

import sqlite3

import sys

from typing import (

    cast,

    Any,

    Callable,

    ClassVar,

    Dict,

    Hashable,

    Iterable,

    Iterator,

    List,

    Mapping,

    MutableMapping,

    NamedTuple,

    Optional,

    Sequence,

    Set,

    TextIO,

    Tuple,

    Type,

    Union,

)

    def __call__(self, context: PostingContext) -> Set[str]:

        raw_value = super().__call__(context)

        seq = raw_value.split() if isinstance(raw_value, str) else ''

        return set(seq)

class RTField(NamedTuple):

    key: str

    parse: Optional[Callable[[str], object]]

    unset_value: Optional[str] = None

    def load(self, rt_ticket: RTResult) -> object:

        value = rt_ticket.get(self.key) if rt_ticket else None

        if not value or value == self.unset_value:

            return None

        elif self.parse is None:

            return value

        else:

            return self.parse(value)

class RTTicket(bc_query_compile.EvalFunction):

    """Look up a field from RT ticket(s) mentioned in metadata documentation"""

    __intypes__ = [str, str, int]

    FIELDS = {key: RTField(key, None) for key in [

        'AdminCc',

        'Cc',

        'Creator',

        'Owner',

        'Queue',

        'Status',

        'Subject',

        'Requestors',

    ]}

    FIELDS.update((key, RTField(key, int, '0')) for key in [

        'numerical_id',

        'FinalPriority',

        'InitialPriority',

        'Priority',

        'TimeEstimated',

        'TimeLeft',

        'TimeWorked',

    ])

    FIELDS.update((key, RTField(key, rtutil.RTDateTime, 'Not set')) for key in [

        'Created',

        'Due',

        'LastUpdated',

        'Resolved',

            'RT_CLIENT': client,

            '_rt_cache': cls._CACHES.setdefault(cache_key, {}),

        })

    def __init__(self, operands: List[bc_query_compile.EvalNode]) -> None:

        if not hasattr(self, 'RT_CLIENT'):

            raise RuntimeError("no RT client available - cannot use rt_ticket()")

        rt_op, meta_op, *rest = operands

        # We have to evaluate the RT and meta keys on each call, because they

        # might themselves be dynamic. In the common case they're constants.

        # In that case, check for typos so we can report an error to the user

        # before execution even begins.

        if isinstance(rt_op, bc_query_compile.EvalConstant):

            self._rt_key(rt_op.value)

        if isinstance(meta_op, bc_query_compile.EvalConstant):

            self._meta_key(meta_op.value)

        if not rest:

            operands.append(bc_query_compile.EvalConstant(sys.maxsize))

        super().__init__(operands, set)

    def _rt_key(self, key: str) -> RTField:

        try:

            return self.FIELDS[key]

        except KeyError:

            raise ValueError(f"unknown RT ticket field {key!r}") from None

    def _meta_key(self, key: str) -> str:

        if key in data.LINK_METADATA:

            return key

        else:

            raise ValueError(f"metadata key {key!r} does not contain documentation links")

    def __call__(self, context: PostingContext) -> Set[object]:

        rt_key: str

        meta_key: str

        limit: int

        rt_key, meta_key, limit = self.eval_args(context)

        rt_field = self._rt_key(rt_key)

        meta_key = self._meta_key(meta_key)

        if limit < 1:

            return set()

        ticket_ids: Set[str] = set()

        for link_s in ContextMeta(context).report_links(meta_key):

            rt_id = rtutil.RT.parse(link_s)

            if rt_id is not None:

                ticket_ids.add(rt_id[0])

                if len(ticket_ids) >= limit:

                    break

#!/usr/bin/env python3

from setuptools import setup

setup(

    name='conservancy_beancount',

    description="Plugin, library, and reports for reading Conservancy's books",

    author='Software Freedom Conservancy',

    author_email='info@sfconservancy.org',

    license='GNU AGPLv3+',

    install_requires=[

        'babel>=2.6',  # Debian:python3-babel

        'beancount>=2.2',  # Debian:beancount

        'GitPython>=2.0',  # Debian:python3-git

        # 1.4.1 crashes when trying to save some documents.

        'odfpy>=1.4.0,!=1.4.1',  # Debian:python3-odf

        'pdfminer.six>=20200101',

        'python-dateutil>=2.7',  # Debian:python3-dateutil

        'PyYAML>=3.0',  # Debian:python3-yaml

        'regex',  # Debian:python3-regex

        'rt>=2.0',

    ],

    setup_requires=[

        'pytest-mypy',

        'pytest-runner',  # Debian:python3-pytest-runner

    ],

    tests_require=[

        'mypy>=0.770',  # Debian:python3-mypy

        'pytest',  # Debian:python3-pytest

    ],

    stdout = stdout_type()

    stderr = io.StringIO()

    returncode = qmod.main(arglist, stdout, stderr, config)

    return returncode, stdout, stderr

def test_rt_ticket_unconfigured():

    with pytest.raises(RuntimeError):

        qmod.RTTicket(const_operands('id', 'rt-id'))

@pytest.mark.parametrize('field_name', ['foo', 'bar'])

def test_rt_ticket_bad_field(ticket_query, field_name):

    with pytest.raises(ValueError):

        ticket_query(const_operands(field_name, 'rt-id'))

@pytest.mark.parametrize('meta_name', ['foo', 'bar'])

def test_rt_ticket_bad_metadata(ticket_query, meta_name):

    with pytest.raises(ValueError):

        ticket_query(const_operands('id', meta_name))

@pytest.mark.parametrize('field_name,meta_name,expected', [

    ('id', 'rt-id', {1}),

    ('Queue', 'approval', {'general'}),

    ('Requestors', 'invoice', {'mx1@example.org', 'requestor2@example.org'}),

    ('Due', 'tax-reporting', {datetime.datetime(2017, 1, 14, 12, 1, 0, tzinfo=UTC)}),

])

def test_rt_ticket_from_txn(ticket_query, field_name, meta_name, expected):

    func = ticket_query(const_operands(field_name, meta_name))

    txn = testutil.Transaction(**{meta_name: 'rt:1'}, postings=[

        ('Assets:Cash', 80),

    ])

    context = RowContext(txn, txn.postings[0])

    assert func(context) == expected

@pytest.mark.parametrize('field_name,meta_name,expected', [

    ('id', 'rt-id', {2}),

    ('Queue', 'approval', {'general'}),

    ('Requestors', 'invoice', {'mx2@example.org', 'requestor2@example.org'}),

    ('Due', 'tax-reporting', {datetime.datetime(2017, 1, 14, 12, 2, 0, tzinfo=UTC)}),

])

def test_rt_ticket_from_post(ticket_query, field_name, meta_name, expected):

    func = ticket_query(const_operands(field_name, meta_name))

    txn = testutil.Transaction(**{meta_name: 'rt:1'}, postings=[

        ('Assets:Cash', 110, {meta_name: 'rt:2/8'}),

    ])

    context = RowContext(txn, txn.postings[0])

    assert func(context) == expected

@pytest.mark.parametrize('field_name,meta_name,expected,on_txn', [

    ('id', 'approval', {1, 2}, True),

    ('Queue', 'check', {'general'}, False),

    ('Requestors', 'invoice', {

        'mx1@example.org',

        'mx2@example.org',

        'requestor2@example.org',

    }, False),

])

def test_rt_ticket_multi_results(ticket_query, field_name, meta_name, expected, on_txn):

    func = ticket_query(const_operands(field_name, meta_name))

    txn = testutil.Transaction(**{'rt-id': 'rt:1'}, postings=[

        ('Assets:Cash', 110, {'rt-id': 'rt:2'}),

    ])

    post = txn.postings[0]

    meta = txn.meta if on_txn else post.meta

    meta[meta_name] = 'rt:1/2 Docs/12.pdf rt:2/8'

    context = RowContext(txn, post)

    assert func(context) == expected

@pytest.mark.parametrize('meta_value,on_txn', testutil.combine_values(

    ['', 'Docs/34.pdf', 'Docs/100.pdf Docs/120.pdf'],

    [True, False],

))

def test_rt_ticket_no_results(ticket_query, meta_value, on_txn):

    func = ticket_query(const_operands('Queue', 'check'))

    txn = testutil.Transaction(**{'rt-id': 'rt:1'}, postings=[

        ('Assets:Cash', 110, {'rt-id': 'rt:2'}),

    ])

    post = txn.postings[0]

    meta = txn.meta if on_txn else post.meta

    meta['check'] = meta_value