1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
// yt - A fully featured command line YouTube client
//
// Copyright (C) 2024 Benedikt Peetz <benedikt.peetz@b-peetz.de>
// SPDX-License-Identifier: GPL-3.0-or-later
//
// This file is part of Yt.
//
// You should have received a copy of the License along with this program.
// If not, see <https://www.gnu.org/licenses/gpl-3.0.txt>.
// This file is taken from: https://github.com/dylanbstorey/pyo3-pylogger/blob/d89e0d6820ebc4f067647e3b74af59dbc4941dd5/src/lib.rs
// It is licensed under the Apache 2.0 License, copyright up to 2024 by Dylan Storey
// It was modified by Benedikt Peetz 2024
use log::{logger, Level, MetadataBuilder, Record};
use pyo3::{
prelude::{PyAnyMethods, PyListMethods, PyModuleMethods},
pyfunction, wrap_pyfunction, Bound, PyAny, PyResult, Python,
};
/// Consume a Python `logging.LogRecord` and emit a Rust `Log` instead.
#[pyfunction]
fn host_log(record: Bound<'_, PyAny>, rust_target: &str) -> PyResult<()> {
let level = record.getattr("levelno")?;
let message = record.getattr("getMessage")?.call0()?.to_string();
let pathname = record.getattr("pathname")?.to_string();
let lineno = record
.getattr("lineno")?
.to_string()
.parse::<u32>()
.expect("This should always be a u32");
let logger_name = record.getattr("name")?.to_string();
let full_target: Option<String> = if logger_name.trim().is_empty() || logger_name == "root" {
None
} else {
// Libraries (ex: tracing_subscriber::filter::Directive) expect rust-style targets like foo::bar,
// and may not deal well with "." as a module separator:
let logger_name = logger_name.replace(".", "::");
Some(format!("{rust_target}::{logger_name}"))
};
let target = full_target.as_deref().unwrap_or(rust_target);
// error
let error_metadata = if level.ge(40u8)? {
MetadataBuilder::new()
.target(target)
.level(Level::Error)
.build()
} else if level.ge(30u8)? {
MetadataBuilder::new()
.target(target)
.level(Level::Warn)
.build()
} else if level.ge(20u8)? {
MetadataBuilder::new()
.target(target)
.level(Level::Info)
.build()
} else if level.ge(10u8)? {
MetadataBuilder::new()
.target(target)
.level(Level::Debug)
.build()
} else {
MetadataBuilder::new()
.target(target)
.level(Level::Trace)
.build()
};
logger().log(
&Record::builder()
.metadata(error_metadata)
.args(format_args!("{}", &message))
.line(Some(lineno))
.file(None)
.module_path(Some(&pathname))
.build(),
);
Ok(())
}
/// Registers the host_log function in rust as the event handler for Python's logging logger
/// This function needs to be called from within a pyo3 context as early as possible to ensure logging messages
/// arrive to the rust consumer.
pub fn setup_logging(py: Python, target: &str) -> PyResult<()> {
let logging = py.import_bound("logging")?;
logging.setattr("host_log", wrap_pyfunction!(host_log, &logging)?)?;
py.run_bound(
format!(
r#"
class HostHandler(Handler):
def __init__(self, level=0):
super().__init__(level=level)
def emit(self, record):
host_log(record,"{}")
oldBasicConfig = basicConfig
def basicConfig(*pargs, **kwargs):
if "handlers" not in kwargs:
kwargs["handlers"] = [HostHandler()]
return oldBasicConfig(*pargs, **kwargs)
"#,
target
)
.as_str(),
Some(&logging.dict()),
None,
)?;
let all = logging.index()?;
all.append("HostHandler")?;
Ok(())
}
|
